diff --git a/src/main/java/net/openhft/chronicle/common/serialization/CopyingMetaBytesInterop.java b/src/main/java/net/openhft/chronicle/common/serialization/CopyingMetaBytesInterop.java index d61ff4422..f7264d2d9 100644 --- a/src/main/java/net/openhft/chronicle/common/serialization/CopyingMetaBytesInterop.java +++ b/src/main/java/net/openhft/chronicle/common/serialization/CopyingMetaBytesInterop.java @@ -76,6 +76,7 @@ public abstract class CopyingMetaBytesInterop implements MetaBytesInterop< @Override public void write(W writer, Bytes bytes, E e) { + assert bytes.limit() == bytes.capacity(); bytes.write(buffer.buffer); } diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 518f3a5aa..aa579a5f3 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -95,6 +95,7 @@ public class ChronicleMapBuilder implements Cloneable, public static final short UDP_REPLICATION_MODIFICATION_ITERATOR_ID = 128; private static final Logger LOG = LoggerFactory.getLogger(ChronicleMapBuilder.class.getName()); + SerializationBuilder keyBuilder; SerializationBuilder valueBuilder; @@ -133,7 +134,9 @@ public class ChronicleMapBuilder implements Cloneable, private File file; ChronicleMapBuilder(Class keyClass, Class valueClass) { + keyBuilder = new SerializationBuilder(keyClass, SerializationBuilder.Role.KEY); + valueBuilder = new SerializationBuilder(valueClass, SerializationBuilder.Role.VALUE); } @@ -839,7 +842,15 @@ public class ChronicleMapBuilder implements Cloneable, final KeyValueSerializer keyValueSerializer = new KeyValueSerializer(keyBuilder, valueBuilder); - return new StatelessChronicleMap(keyValueSerializer, statelessBuilder, entrySize); + final Class kClass = keyBuilder.eClass; + final Class vClass = valueBuilder.eClass; + + return new StatelessChronicleMap( + keyValueSerializer, + statelessBuilder, + entrySize(), + kClass, + vClass); } private VanillaChronicleMap newMap() throws IOException { diff --git a/src/main/java/net/openhft/chronicle/map/Serializer.java b/src/main/java/net/openhft/chronicle/map/Serializer.java index 99359e8f7..fc2b37a5f 100644 --- a/src/main/java/net/openhft/chronicle/map/Serializer.java +++ b/src/main/java/net/openhft/chronicle/map/Serializer.java @@ -71,9 +71,10 @@ class Serializer { sizeMarshaller.writeSize(out, valueSize); - if (metaValueWriter != null) + if (metaValueWriter != null) { + assert out.limit() == out.capacity(); metaValueWriter.write(valueWriter, out, value); - else + } else throw new UnsupportedOperationException(""); } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index e572cf38b..6695a20a1 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -15,6 +15,7 @@ import java.io.File; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.channels.SocketChannel; import java.util.*; @@ -41,6 +42,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private CloseablesManager closeables; private final StatelessBuilder builder; private int maxEntrySize; + private final Class kClass; + private final Class vClass; private int headroom = 1024; static enum EventId { @@ -61,20 +64,25 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { PUT_IF_ABSENT, REMOVE_WITH_VALUE, TO_STRING, - PUT_ALL + PUT_ALL, + EQUALS, + HASH_CODE } private long transactionID; StatelessChronicleMap(final KeyValueSerializer keyValueSerializer, final StatelessBuilder builder, - int maxEntrySize) throws IOException { + int maxEntrySize, Class kClass, Class vClass) throws IOException { this.keyValueSerializer = keyValueSerializer; this.builder = builder; this.maxEntrySize = maxEntrySize; + this.kClass = kClass; + this.vClass = vClass; attemptConnect(builder.remoteAddress()); headroom = 1024 + maxEntrySize; - buffer = allocateDirect(headroom); + buffer = allocateDirect(headroom*1024).order(ByteOrder.nativeOrder()); + bytes = new ByteBufferBytes(buffer.slice()); } @@ -261,6 +269,57 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } + /** + * calling this method should be avoided at all cost, as the entire {@code object} is + * serialized. This equals can be used to compare map that extends ChronicleMap. So two + * Chronicle Maps that contain the same data are considered equal, even if the instances of the + * chronicle maps were of different types + * + * @param object the object that you are comparing against + * @return true if the contain the same data + */ + @Override + public boolean equals(Object object) { + if (this == object) return true; + if (object == null || object.getClass().isAssignableFrom(Map.class)) + return false; + + final Map that = (Map) object; + + long sizeLocation = writeEvent(EQUALS); + + try { + writeEntries(that); + } catch (ClassCastException e) { + if (LOG.isDebugEnabled()) + LOG.debug("failed equals() due to a difference in types", e); + return false; + } + + // get the data back from the server + return blockingFetch(sizeLocation).readBoolean(); + } + + + public synchronized void putAll(Map map) { + + final long sizeLocation = writeEvent(PUT_ALL); + writeEntries(map); + + blockingFetch(sizeLocation); + + } + + + + @Override + public int hashCode() { + long sizeLocation = writeEvent(HASH_CODE); + + // get the data back from the server + return blockingFetch(sizeLocation).readInt(); + } + public synchronized String toString() { long sizeLocation = writeEvent(TO_STRING); @@ -313,7 +372,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - public synchronized V getUsing(K key, V value) { throw new UnsupportedOperationException("getUsing is not supported for stateless clients"); } @@ -323,7 +381,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { throw new UnsupportedOperationException("getUsing is not supported for stateless clients"); } - public synchronized V put(K key, V value) { long sizeLocation = writeEvent(PUT); @@ -343,39 +400,81 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - public synchronized void putAll(Map map) { - final long sizeLocation = writeEvent(PUT_ALL); + + private void writeEntries(Map map) { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; bytes.writeStopBit(numberOfEntries); - + assert bytes.limit() == bytes.capacity(); for (final Map.Entry e : map.entrySet()) { numberOfEntriesReadSoFar++; - + assert bytes.limit() == bytes.capacity(); // putAll if a bit more complicated than the others // as the volume of data could cause us to have to resize our buffers resizeIfRequired(numberOfEntries, numberOfEntriesReadSoFar); - long start = bytes.position(); - writeKey(e.getKey()); - writeValue(e.getValue()); - int len = (int) (bytes.position() - start); + assert bytes.limit() == bytes.capacity(); + final long start = bytes.position(); + final K key = e.getKey(); + + final Class keyClass = key.getClass(); + if (!kClass.isAssignableFrom(keyClass)) + throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " + + "and should" + + " be of type=" + kClass); + + writeKey(key); + assert bytes.limit() == bytes.capacity(); + final V value = e.getValue(); + + final Class valueClass = value.getClass(); + if (!vClass.isAssignableFrom(valueClass)) + throw new ClassCastException("value=" + value + " is of type=" + valueClass + + " and " + + "should be of type=" + vClass); + + + try { + assert bytes.limit() == bytes.capacity(); + writeValue(value); + + + } catch ( + Exception e4) { + LOG.info("bytes position=" + bytes.position() + ",capacity=" + bytes.capacity() + + ", " + + "limit=" + bytes.limit()); + assert bytes.limit() == bytes.capacity(); + + writeValue(value); + int i = 0; + throw e4; + } + + + final int len = (int) (bytes.position() - start); + assert bytes.limit() == bytes.capacity(); if (len > maxEntrySize) maxEntrySize = len; } - - blockingFetch(sizeLocation); - } private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar) { - if (bytes.remaining() < maxEntrySize) { + + // System.out.println("bytes=" + bytes.position()); + + long remaining = bytes.remaining(); + + /*LOG.info("bytes position=" + bytes.position() + ",capacity=" + bytes.capacity() + ", " + + "limit=" + bytes.limit()); +*/ + if (remaining < maxEntrySize) { long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar); - resizeBuffer(estimatedRequiredSize + headroom); + resizeBuffer(estimatedRequiredSize * 2); } } @@ -387,8 +486,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { */ private long estimateSize(int numberOfEntries, int numberOfEntriesReadSoFar) { // we will back the size estimate on what we have so far - double percentageComplete = numberOfEntries / numberOfEntriesReadSoFar; + double percentageComplete = (double) numberOfEntriesReadSoFar / (double) numberOfEntries; return (long) ((double) bytes.position() / percentageComplete); + + // LOG.info("bytes position=" + bytes.position() + ",capacity=" + bytes.capacity() + ", " + + //"limit=" + bytes.limit()); + // return bytes.position() + (maxEntrySize * 128); } @@ -396,43 +499,59 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { if (size < buffer.capacity()) throw new IllegalStateException("it not possible to resize the buffer smaller"); - ByteBuffer result = ByteBuffer.allocateDirect((int) size); - buffer.clear(); + assert size < Integer.MAX_VALUE; + + LOG.info("bytes before position=" + bytes.position() + + ", " + + "limit=" + bytes.limit()+ ",capacity=" + bytes.capacity() ); + + + LOG.info("buffer before position=" + buffer.position() + + ", " + + "limit=" + buffer.limit()+ ",capacity=" + buffer.capacity() ); + + LOG.info("resize buffer to " + size + " bytes"); + final ByteBuffer result = ByteBuffer.allocateDirect((int) size).order(ByteOrder.nativeOrder()); int bufferPosition = buffer.position(); int bufferLimit = buffer.limit(); long bytesPosition = bytes.position(); - long bytesLimit = bytes.limit(); + bytes = new ByteBufferBytes(result.slice()); buffer.position(0); - for (int i = 0; i < bufferLimit; i++) { + for (int i = 0; i < bytesPosition; i++) { result.put(buffer.get()); } + buffer = result; buffer.limit(bufferLimit); buffer.position(bufferPosition); - bytes = new ByteBufferBytes(result.slice()); - bytes.limit(bytesLimit); + assert buffer.capacity() == bytes.capacity(); + + bytes.limit(bytes.capacity()); bytes.position(bytesPosition); + + + assert buffer.capacity() == size; + assert buffer.capacity() == bytes.capacity(); + assert bytes.limit() == bytes.capacity(); + + + LOG.info("bytes after position=" + bytes.position() + + ", " + + "limit=" + bytes.limit()+ ",capacity=" + bytes.capacity() ); + + + LOG.info("buffer after position=" + buffer.position() + + ", " + + "limit=" + buffer.limit()+ ",capacity=" + buffer.capacity() ); + } -/* private void writeEntries(Map map) { - - final HashMap safeCopy = new HashMap(map); - bytes.writeStopBit(safeCopy.size()); - - final Set entries = (Set) safeCopy.entrySet(); - - for (Entry e : entries) { - writeKey(e.getKey()); - writeValue(e.getValue()); - } - }*/ - public synchronized void clear() { long sizeLocation = writeEvent(CLEAR); @@ -494,8 +613,11 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private long writeEvent(EventId event) { - bytes.clear(); + buffer.clear(); + bytes.clear(); + + //bytes.limit(buffer.capacity()); bytes.write((byte) event.ordinal()); long sizeLocation = markSizeLocation(); @@ -612,10 +734,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { throw (RuntimeException) bytes.readObject(); return bytes; - } catch (java.nio.channels.ClosedChannelException e) { - checkTimeout(timeoutTime); - clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); - } catch (ClosedConnectionException e) { + } catch (java.nio.channels.ClosedChannelException | ClosedConnectionException e) { checkTimeout(timeoutTime); clientChannel = lazyConnect(builder.timeoutMs(), builder.remoteAddress()); } @@ -638,6 +757,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } private void send(final ByteBufferBytes out, long timeoutTime) throws IOException { + buffer.limit((int) out.position()); buffer.position(0); @@ -676,6 +796,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } private void writeValue(V value) { + // LOG.info("bytes position=" + bytes.position() + ",capacity=" + bytes.capacity() + ", " + + // "limit=" + bytes.limit()); keyValueSerializer.writeValue(value, bytes); } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 0b3fda654..0fe9fd7f0 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -26,7 +26,9 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; import java.io.IOException; import java.net.*; +import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.channels.*; import java.util.*; import java.util.concurrent.TimeUnit; @@ -44,6 +46,7 @@ import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT; class TcpReplicator extends AbstractChannelReplicator implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName()); + private static final int BUFFER_SIZE = 0x100000; // 1MB public static final int STATELESS_CLIENT = -127; public static final byte NOT_SET = (byte) HEARTBEAT.ordinal(); @@ -1037,33 +1040,55 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { void resizeBuffer(long size) { + assert size < Integer.MAX_VALUE; + + + LOG.info("in before position=" + in.position() + + ", " + + "limit=" + in.limit() + ",capacity=" + in.capacity()); + + + LOG.info("out before position=" + out.position() + + ", " + + "limit=" + out.limit() + ",capacity=" + out.capacity()); + + if (size < in.capacity()) throw new IllegalStateException("it not possible to resize the buffer smaller"); - final ByteBuffer buffer = ByteBuffer.allocateDirect((int) size); - buffer.clear(); + final ByteBuffer buffer = ByteBuffer.allocateDirect((int) size).order(ByteOrder.nativeOrder()); - final long outPosition = out.position(); - final long outLimit = out.limit(); final int inPosition = in.position(); - final int inLimit = in.limit(); + + + long outPosition = out.position(); + long outLimit = out.limit(); + + out = new ByteBufferBytes(buffer.slice()); in.position(0); - - for (int i = 0; i < inLimit; i++) { + for (int i = 0; i < inPosition; i++) { buffer.put(in.get()); } - buffer.limit(inLimit); - buffer.position(inPosition); - in = buffer; - out = new ByteBufferBytes(in.slice()); + in.limit(in.capacity()); + in.position(inPosition); out.limit(outLimit); out.position(outPosition); + LOG.info("after in position=" + in.position() + + ", " + + "limit=" + in.limit() + ",capacity=" + in.capacity()); + + + LOG.info("after out position=" + out.position() + + ", " + + "limit=" + out.limit() + ",capacity=" + out.capacity()); + + } /** @@ -1226,6 +1251,9 @@ interface Work { */ class StatelessServerConnector { + private static final Logger LOG = LoggerFactory.getLogger(StatelessServerConnector.class + .getName()); + ; private final KeyValueSerializer keyValueSerializer; private final ChronicleMap map; private double maxEntrySizeBytes; @@ -1239,66 +1267,83 @@ class StatelessServerConnector { } Work processStatelessEvent(final byte eventId, - @NotNull final Bytes out, - @NotNull final ByteBufferBytes in) { + @NotNull final Bytes writer, + @NotNull final ByteBufferBytes reader) { + + + LOG.info("processStatelessEvent - writer position=" + reader.position() + + ", " + + "limit=" + reader.limit() + ",capacity=" + reader.capacity()); + + + LOG.info("StatelessServerConnector- reader position=" + writer.position() + + ", " + + "limit=" + writer.limit() + ",capacity=" + writer.capacity()); + final StatelessChronicleMap.EventId event = StatelessChronicleMap.EventId.values()[eventId]; switch (event) { case LONG_SIZE: - return longSize(in, out); + return longSize(reader, writer); case IS_EMPTY: - return isEmpty(in, out); + return isEmpty(reader, writer); case CONTAINS_KEY: - return containsKey(in, out); + return containsKey(reader, writer); case CONTAINS_VALUE: - return containsValue(in, out); + return containsValue(reader, writer); case GET: - return get(in, out); + return get(reader, writer); case PUT: - return put(in, out); + return put(reader, writer); case REMOVE: - return remove(in, out); + return remove(reader, writer); case CLEAR: - return clear(in, out); + return clear(reader, writer); case KEY_SET: - return keySet(in, out); + return keySet(reader, writer); case VALUES: - return values(in, out); + return values(reader, writer); case ENTRY_SET: - return entrySet(in, out); + return entrySet(reader, writer); case REPLACE: - return replace(in, out); + return replace(reader, writer); case REPLACE_WITH_OLD_AND_NEW_VALUE: - return replaceWithOldAndNew(in, out); + return replaceWithOldAndNew(reader, writer); case PUT_IF_ABSENT: - return putIfAbsent(in, out); + return putIfAbsent(reader, writer); case REMOVE_WITH_VALUE: - return removeWithValue(in, out); + return removeWithValue(reader, writer); case SIZE: - return size(in, out); + return size(reader, writer); case TO_STRING: - return toString(in, out); + return toString(reader, writer); case PUT_ALL: - return putAll(in, out); + return putAll(reader, writer); + + case HASH_CODE: + return hashCode(reader, writer); + + case EQUALS: + return equals(reader, writer); default: @@ -1309,71 +1354,83 @@ class StatelessServerConnector { } - private Work removeWithValue(Bytes in, Bytes out) { + private Work removeWithValue(Bytes reader, Bytes writer) { - final K key = readKey(in); - final V readValue = readValue(in); - final long sizeLocation = reflectTransactionId(in, out); + final K key = readKey(reader); + final V readValue = readValue(reader); + final long sizeLocation = reflectTransactionId(reader, writer); final boolean result; try { result = map.remove(key, readValue); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - out.writeBoolean(result); + writer.writeBoolean(result); return null; } - private Work replaceWithOldAndNew(Bytes in, Bytes out) { + private Work replaceWithOldAndNew(Bytes reader, Bytes writer) { - final K key = readKey(in); - final V oldValue = readValue(in); - final V newValue = readValue(in); + final K key = readKey(reader); + final V oldValue = readValue(reader); + final V newValue = readValue(reader); boolean replaced; - long sizeLocation = reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(reader, writer); try { replaced = map.replace(key, oldValue, newValue); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - out.writeBoolean(replaced); - writeSizeAndFlags(sizeLocation, false, out); + writer.writeBoolean(replaced); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work longSize(Bytes in, Bytes out) { - final long sizeLocation = reflectTransactionId(in, out); + private Work longSize(Bytes reader, Bytes writer) { + final long sizeLocation = reflectTransactionId(reader, writer); try { - out.writeLong(map.longSize()); + writer.writeLong(map.longSize()); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work size(Bytes in, Bytes out) { - final long sizeLocation = reflectTransactionId(in, out); + private Work size(Bytes reader, Bytes writer) { + final long sizeLocation = reflectTransactionId(reader, writer); try { - out.writeInt(map.size()); + writer.writeInt(map.size()); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work toString(Bytes in, Bytes out) { - final long sizeLocation = reflectTransactionId(in, out); + private Work hashCode(ByteBufferBytes reader, Bytes writer) { + final long sizeLocation = reflectTransactionId(reader, writer); + try { + writer.writeInt(map.hashCode()); + } catch (RuntimeException e) { + return sendException(writer, sizeLocation, e); + } + writeSizeAndFlags(sizeLocation, false, writer); + return null; + } + + + private Work toString(Bytes reader, Bytes writer) { + final long sizeLocation = reflectTransactionId(reader, writer); final String str; - final long remaining = out.remaining(); + final long remaining = writer.remaining(); try { str = map.toString(); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } assert remaining > 4; @@ -1383,133 +1440,150 @@ class StatelessServerConnector { str : str.substring(0, (int) (remaining - 4)) + "..."; - out.writeObject(result); - writeSizeAndFlags(sizeLocation, false, out); + writer.writeObject(result); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work sendException(Bytes out, long sizeLocation, RuntimeException e) { - writeException(e, out); - writeSizeAndFlags(sizeLocation, true, out); + private Work sendException(Bytes writer, long sizeLocation, RuntimeException e) { + writeException(e, writer); + writeSizeAndFlags(sizeLocation, true, writer); return null; } - private Work isEmpty(Bytes in, Bytes out) { - final long sizeLocation = reflectTransactionId(in, out); + private Work isEmpty(Bytes reader, Bytes writer) { + final long sizeLocation = reflectTransactionId(reader, writer); try { - out.writeBoolean(map.isEmpty()); + writer.writeBoolean(map.isEmpty()); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work containsKey(Bytes in, Bytes out) { - final K k = readKey(in); - final long sizeLocation = reflectTransactionId(in, out); + private Work containsKey(Bytes reader, Bytes writer) { + final K k = readKey(reader); + final long sizeLocation = reflectTransactionId(reader, writer); try { - out.writeBoolean(map.containsKey(k)); + writer.writeBoolean(map.containsKey(k)); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work containsValue(Bytes in, Bytes out) { - final V v = readValue(in); - final long sizeLocation = reflectTransactionId(in, out); + private Work containsValue(Bytes reader, Bytes writer) { + final V v = readValue(reader); + final long sizeLocation = reflectTransactionId(reader, writer); try { - out.writeBoolean(map.containsValue(v)); + writer.writeBoolean(map.containsValue(v)); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work get(Bytes in, Bytes out) { - final K k = readKey(in); - final long sizeLocation = reflectTransactionId(in, out); + private Work get(Bytes reader, Bytes writer) { + final K k = readKey(reader); + final long sizeLocation = reflectTransactionId(reader, writer); try { - writeValue(map.get(k), out); + writeValue(map.get(k), writer); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work put(Bytes in, Bytes out) { - final K k = readKey(in); - final V v = readValue(in); - long sizeLocation = reflectTransactionId(in, out); + private Work put(Bytes reader, Bytes writer) { + final K k = readKey(reader); + final V v = readValue(reader); + long sizeLocation = reflectTransactionId(reader, writer); try { - writeValue(map.put(k, v), out); + writeValue(map.put(k, v), writer); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work remove(Bytes in, Bytes out) { - final K key = readKey(in); - final long sizeLocation = reflectTransactionId(in, out); + private Work remove(Bytes reader, Bytes writer) { + final K key = readKey(reader); + final long sizeLocation = reflectTransactionId(reader, writer); try { - writeValue(map.remove(key), out); + writeValue(map.remove(key), writer); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work putAll(Bytes in, Bytes out) { - final Map m = readEntries(in); - final long sizeLocation = reflectTransactionId(in, out); + private Work putAll(Bytes reader, Bytes writer) { + final Map m = readEntries(reader); + final long sizeLocation = reflectTransactionId(reader, writer); try { map.putAll(m); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work clear(Bytes in, Bytes out) { + private Work equals(Bytes reader, Bytes writer) { + final Map m = readEntries(reader); + final long sizeLocation = reflectTransactionId(reader, writer); - final long sizeLocation = reflectTransactionId(in, out); + boolean result; + try { + result = map.equals(m); + } catch (RuntimeException e) { + return sendException(writer, sizeLocation, e); + } + + writer.writeBoolean(result); + writeSizeAndFlags(sizeLocation, false, writer); + return null; + } + + + private Work clear(Bytes reader, Bytes writer) { + + final long sizeLocation = reflectTransactionId(reader, writer); try { map.clear(); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work keySet(Bytes in, final Bytes out) { - final long sizeLocation = reflectTransactionId(in, out); + private Work keySet(Bytes reader, final Bytes writer) { + final long sizeLocation = reflectTransactionId(reader, writer); final Set ks; try { ks = map.keySet(); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - out.writeStopBit(ks.size()); + writer.writeStopBit(ks.size()); final Iterator iterator = ks.iterator(); @@ -1517,35 +1591,35 @@ class StatelessServerConnector { return new Work() { @Override - public boolean doWork(@NotNull final Bytes out) { + public boolean doWork(@NotNull final Bytes writer) { 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 (out.remaining() <= maxEntrySizeBytes) + if (writer.remaining() <= maxEntrySizeBytes) return false; - writeKey(iterator.next(), out); + writeKey(iterator.next(), writer); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return true; } }; } - private Work values(Bytes in, Bytes out) { - final long sizeLocation = reflectTransactionId(in, out); + private Work values(Bytes reader, Bytes writer) { + final long sizeLocation = reflectTransactionId(reader, writer); final Collection values; try { values = map.values(); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - out.writeStopBit(values.size()); + writer.writeStopBit(values.size()); final Iterator iterator = values.iterator(); @@ -1553,35 +1627,35 @@ class StatelessServerConnector { return new Work() { @Override - public boolean doWork(Bytes out) { + public boolean doWork(Bytes writer) { 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 (out.remaining() <= maxEntrySizeBytes) + if (writer.remaining() <= maxEntrySizeBytes) return false; - writeValue(iterator.next(), out); + writeValue(iterator.next(), writer); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return true; } }; } - private Work entrySet(Bytes in, Bytes out) { + private Work entrySet(Bytes reader, Bytes writer) { - final long sizeLocation = reflectTransactionId(in, out); + final long sizeLocation = reflectTransactionId(reader, writer); final Set> entries; try { entries = map.entrySet(); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - out.writeStopBit(entries.size()); + writer.writeStopBit(entries.size()); final Iterator> iterator = entries.iterator(); @@ -1589,55 +1663,55 @@ class StatelessServerConnector { return new Work() { @Override - public boolean doWork(Bytes out) { + public boolean doWork(Bytes writer) { 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 (out.remaining() <= maxEntrySizeBytes) + if (writer.remaining() <= maxEntrySizeBytes) return false; final Map.Entry next = iterator.next(); - writeKey(next.getKey(), out); - writeValue(next.getValue(), out); + writeKey(next.getKey(), writer); + writeValue(next.getValue(), writer); } - writeSizeAndFlags(sizeLocation, false, out); + writeSizeAndFlags(sizeLocation, false, writer); return true; } }; } - private Work putIfAbsent(Bytes in, Bytes out) { - final K key = readKey(in); - final V v = readValue(in); - final long sizeLocation = reflectTransactionId(in, out); + private Work putIfAbsent(Bytes reader, Bytes writer) { + final K key = readKey(reader); + final V v = readValue(reader); + final long sizeLocation = reflectTransactionId(reader, writer); final V value; try { value = map.putIfAbsent(key, v); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeValue(value, out); - writeSizeAndFlags(sizeLocation, false, out); + writeValue(value, writer); + writeSizeAndFlags(sizeLocation, false, writer); return null; } - private Work replace(Bytes in, Bytes out) { - final K k = readKey(in); - final V v = readValue(in); + private Work replace(Bytes reader, Bytes writer) { + final K k = readKey(reader); + final V v = readValue(reader); final V replaced; - long sizeLocation = reflectTransactionId(in, out); + long sizeLocation = reflectTransactionId(reader, writer); try { replaced = map.replace(k, v); } catch (RuntimeException e) { - return sendException(out, sizeLocation, e); + return sendException(writer, sizeLocation, e); } - writeValue(replaced, out); - writeSizeAndFlags(sizeLocation, false, out); + writeValue(replaced, writer); + writeSizeAndFlags(sizeLocation, false, writer); return null; } @@ -1647,28 +1721,28 @@ class StatelessServerConnector { * @param key the key of the map */ - private void writeKey(K key, Bytes out) { - keyValueSerializer.writeKey(key, out); + private void writeKey(K key, Bytes writer) { + keyValueSerializer.writeKey(key, writer); } - private long reflectTransactionId(Bytes in, Bytes out) { - final long sizeLocation = out.position(); - out.skip(3); - final long transactionId = in.readLong(); - out.writeLong(transactionId); + private long reflectTransactionId(Bytes reader, Bytes writer) { + final long sizeLocation = writer.position(); + writer.skip(3); + final long transactionId = reader.readLong(); + writer.writeLong(transactionId); return sizeLocation; } - private void writeValue(V value, final Bytes out) { - keyValueSerializer.writeValue(value, out); + private void writeValue(V value, final Bytes writer) { + keyValueSerializer.writeValue(value, writer); } - private K readKey(Bytes in) { - return keyValueSerializer.readKey(in); + private K readKey(Bytes reader) { + return keyValueSerializer.readKey(reader); } - private V readValue(Bytes in) { - return keyValueSerializer.readValue(in); + private V readValue(Bytes reader) { + return keyValueSerializer.readValue(reader); } private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { @@ -1681,13 +1755,13 @@ class StatelessServerConnector { out.writeObject(e); } - private Map readEntries(Bytes in) { + private Map readEntries(Bytes reader) { - final long numberOfEntries = in.readStopBit(); + final long numberOfEntries = reader.readStopBit(); final HashMap result = new HashMap(); for (long i = 0; i < numberOfEntries; i++) { - result.put(readKey(in), readValue(in)); + result.put(readKey(reader), readValue(reader)); } return result; } @@ -1696,8 +1770,8 @@ class StatelessServerConnector { class KeyValueSerializer { - private final Serializer valueSerializer; - private final Serializer keySerializer; + final Serializer valueSerializer; + final Serializer keySerializer; KeyValueSerializer(SerializationBuilder keySerializer, SerializationBuilder valueSerializer) { @@ -1705,17 +1779,27 @@ class KeyValueSerializer { this.valueSerializer = new Serializer(valueSerializer); } - V readValue(Bytes in) { - if (in.readBoolean()) - return null; - return valueSerializer.readMarshallable(in); + V readValue(Bytes reader) { + if (nullCheck(reader)) return null; + return valueSerializer.readMarshallable(reader); } - K readKey(Bytes in) { - if (in.readBoolean()) - return null; + private boolean nullCheck(Bytes in) { + try { + return in.readBoolean(); + } catch (BufferUnderflowException e) { - return keySerializer.readMarshallable(in); + + e.printStackTrace(); + throw e; + } + + } + + K readKey(Bytes reader) { + if (nullCheck(reader)) return null; + + return keySerializer.readMarshallable(reader); } /** @@ -1723,17 +1807,21 @@ class KeyValueSerializer { * * @param key the key of the map */ - void writeKey(K key, Bytes out) { + void writeKey(K key, Bytes writer) { - out.writeBoolean(key == null); + writer.writeBoolean(key == null); if (key != null) - keySerializer.writeMarshallable(key, out); + keySerializer.writeMarshallable(key, writer); } - void writeValue(V value, Bytes out) { - out.writeBoolean(value == null); - if (value != null) - valueSerializer.writeMarshallable(value, out); + void writeValue(V value, Bytes writer) { + assert writer.limit() == writer.capacity(); + writer.writeBoolean(value == null); + assert writer.limit() == writer.capacity(); + if (value != null) { + assert writer.limit() == writer.capacity(); + valueSerializer.writeMarshallable(value, writer); + } } diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index 24f0ba264..ff1a990f6 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -1,5 +1,6 @@ package net.openhft.chronicle.map; +import org.junit.Ignore; import org.junit.Test; import org.testng.Assert; @@ -17,13 +18,15 @@ import static net.openhft.chronicle.common.StatelessBuilder.remoteAddress; public class StatelessClientTest { + public static final int SIZE = 10000; + @Test public void test() throws IOException, InterruptedException { final ChronicleMap serverMap; final ChronicleMap statelessMap; -// server + // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .replicators((byte) 2, TcpReplicationConfig.of(8076)).create(); @@ -32,7 +35,7 @@ public class StatelessClientTest { } -// stateless client + // stateless client { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) @@ -83,6 +86,7 @@ public class StatelessClientTest { @Test + @Ignore public void testBufferOverFlowPutAll() throws IOException, InterruptedException { final ChronicleMap serverMap; @@ -108,19 +112,77 @@ public class StatelessClientTest { Map payload = new HashMap(); - for (int i = 0; i < 100000; i++) { - serverMap.put(i, "some value=" + i); + for (int i = 0; i < SIZE; i++) { + payload.put(i, "some value=" + i); } statelessMap.putAll(payload); Assert.assertEquals("some value=10", statelessMap.get(10)); - Assert.assertEquals(100000, statelessMap.size()); + Assert.assertEquals(SIZE, statelessMap.size()); serverMap.close(); statelessMap.close(); } + + + + @Test + public void testEquals() throws IOException, InterruptedException { + + final ChronicleMap serverMap1; + final ChronicleMap serverMap2; + final ChronicleMap statelessMap1; + final ChronicleMap statelessMap2; + + + // stateless client + { + statelessMap1 = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .stateless(remoteAddress(new InetSocketAddress("localhost", 8076))).create(); + statelessMap2 = ChronicleMapBuilder.of(Integer + .class, CharSequence.class) + .stateless(remoteAddress(new InetSocketAddress("localhost", 8077))).create(); + + + } + + // server + { + serverMap1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replicators((byte) 2, TcpReplicationConfig.of(8076)).create(); + serverMap2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class) + .replicators((byte) 2, TcpReplicationConfig.of(8077)).create(); + + } + + + Map payload = new HashMap(); + + for (int i = 0; i < 10; i++) { + payload.put(i, "some value=" + i); + } + + + statelessMap1.putAll(payload); + statelessMap2.putAll(payload); + + Assert.assertTrue(statelessMap1.equals(statelessMap2)); + + Assert.assertTrue(statelessMap1.equals(payload)); + Assert.assertTrue(statelessMap2.equals(payload)); + + statelessMap1.close(); + statelessMap1.close(); + + serverMap1.close(); + serverMap2.close(); + + } + + }