From cbe6253ca7b43391ee69d774539f0256ebbab50d Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Fri, 24 Oct 2014 14:54:17 +0100 Subject: [PATCH] HCOLL-167 added support for larger entry sizes --- .../map/AbstractChannelReplicator.java | 21 ++++---- .../openhft/chronicle/map/NodeDiscovery.java | 32 ++++++------ .../chronicle/map/StatelessChronicleMap.java | 52 +++++++++---------- .../openhft/chronicle/map/TcpReplicator.java | 14 ++--- .../openhft/chronicle/map/UdpReplicator.java | 12 ++--- .../chronicle/map/StatelessClientTest.java | 9 ---- 6 files changed, 67 insertions(+), 73 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 9cae22389..9da5c012e 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -36,7 +36,6 @@ import static java.lang.Math.round; import static java.nio.channels.SelectionKey.OP_WRITE; import static java.util.concurrent.TimeUnit.DAYS; import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static net.openhft.chronicle.map.ReplicatedChronicleMap.MAX_UNSIGNED_SHORT; /** * @author Rob Austin. @@ -44,7 +43,8 @@ import static net.openhft.chronicle.map.ReplicatedChronicleMap.MAX_UNSIGNED_SHOR abstract class AbstractChannelReplicator implements Closeable { public static final int BITS_IN_A_BYTE = 8; - public static final int SIZE_OF_SHORT = 2; + + public static final int SIZE_OF_SIZE = 4; private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class); final Selector selector; final CloseablesManager closeables = new CloseablesManager(); @@ -143,7 +143,7 @@ abstract class AbstractChannelReplicator implements Closeable { // core dump. executorService.awaitTermination(5, TimeUnit.SECONDS); } catch (InterruptedException e) { - LOG.error("",e); + LOG.error("", e); } } @@ -304,7 +304,7 @@ abstract class AbstractChannelReplicator implements Closeable { long sizeLocation = in.position(); // this is where we will store the size of the entry - in.skip(SIZE_OF_SHORT); + in.skip(SIZE_OF_SIZE); final long start = in.position(); externalizable.writeExternalEntry(entry, in, chronicleId); @@ -316,14 +316,17 @@ abstract class AbstractChannelReplicator implements Closeable { // write the length of the entry, just before the start, so when we read it back // we read the length of the entry first and hence know how many preceding writer to read - final int entrySize = (int) (in.position() - start); - if (entrySize > MAX_UNSIGNED_SHORT) + final long entrySize = (int) (in.position() - start); + if (entrySize > Integer.MAX_VALUE) throw new IllegalStateException("entry too large, the entry size=" + entrySize + ", " + - "entries are limited to a size of " + MAX_UNSIGNED_SHORT); + "entries are limited to a size of " + Integer.MAX_VALUE); + + int entrySize1 = (int) entrySize; if (LOG.isDebugEnabled()) - LOG.debug("sending entry of entrySize=" + entrySize); - in.writeUnsignedShort(sizeLocation, entrySize); + LOG.debug("sending entry of entrySize=" + entrySize1); + + in.writeInt(sizeLocation, entrySize1); return true; } diff --git a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java index ad2723a86..a811bf52a 100644 --- a/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java +++ b/src/main/java/net/openhft/chronicle/map/NodeDiscovery.java @@ -238,8 +238,8 @@ public class NodeDiscovery { /** - * bitwise OR's the two bit sets or put another way, merges the source bitset into the destination bitset - * and returns the destination + * bitwise OR's the two bit sets or put another way, merges the source bitset into the + * destination bitset and returns the destination * * @param source * @param destination @@ -305,8 +305,8 @@ public class NodeDiscovery { } /** - * Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote nodes - * just by name, this functionality requires UDP + * Broadcast the nodes host ports and identifiers over UDP, to make it easy to join a grid of remote + * nodes just by name, this functionality requires UDP * * @author Rob Austin. */ @@ -394,17 +394,17 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator { final int bytesRead = in.position(); - if (bytesRead < SIZE_OF_SHORT + SIZE_OF_SHORT) + if (bytesRead < SIZE_OF_SIZE + SIZE_OF_SIZE) return; out.limit(in.position()); - final short invertedSize = out.readShort(); - final int size = out.readUnsignedShort(); + final int invertedSize = out.readInt(); + final int size = out.readInt(); // check the the first 4 bytes are the inverted len followed by the len // we do this to check that this is a valid start of entry, otherwise we throw it away - if (((short) ~size) != invertedSize) + if (~size != invertedSize) return; if (out.remaining() != size) @@ -450,10 +450,10 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator { in.clear(); // skip the size inverted - in.skip(SIZE_OF_SHORT); + in.skip(SIZE_OF_SIZE); // skip the size - in.skip(SIZE_OF_SHORT); + in.skip(SIZE_OF_SIZE); long start = in.position(); @@ -463,8 +463,8 @@ class NodeDiscoveryBroadcaster extends UdpChannelReplicator { long size = in.position() - start; // we'll write the size inverted at the start - in.writeShort(0, ~((int) size)); - in.writeUnsignedShort(SIZE_OF_SHORT, (int) size); + in.writeInt(0, ~((int) size)); + in.writeInt(SIZE_OF_SIZE, (int) size); out.limit((int) in.position()); @@ -717,8 +717,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { } /** - * this is used to tell nodes that are connecting to us which host and ports are in our grid, along with - * all the identifiers. + * this is used to tell nodes that are connecting to us which host and ports are in our grid, + * along with all the identifiers. */ @Override public void writeMarshallable(@net.openhft.lang.model.constraints.NotNull Bytes out) { @@ -908,8 +908,8 @@ class DiscoveryNodeBytesMarshallable implements BytesMarshallable { /** - * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the host:port - * and perhaps even proposed identifier of the node that sent it. + * sends a bootstrap message to the other nodes in the grid, the bootstrap message contains the + * host:port and perhaps even proposed identifier of the node that sent it. * * @param proposedNodes */ diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index 6695a20a1..74bbc63d2 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 @@ import java.nio.channels.SocketChannel; import java.util.*; import static java.nio.ByteBuffer.allocateDirect; +import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*; /** @@ -80,8 +81,8 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { this.kClass = kClass; this.vClass = vClass; attemptConnect(builder.remoteAddress()); - headroom = 1024 + maxEntrySize; - buffer = allocateDirect(headroom*1024).order(ByteOrder.nativeOrder()); + // headroom = 1024 + maxEntrySize; + buffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder()); bytes = new ByteBufferBytes(buffer.slice()); @@ -281,7 +282,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { @Override public boolean equals(Object object) { if (this == object) return true; - if (object == null || object.getClass().isAssignableFrom(Map.class)) + if (object == null || object.getClass().isAssignableFrom(Map.class)) return false; final Map that = (Map) object; @@ -311,7 +312,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - @Override public int hashCode() { long sizeLocation = writeEvent(HASH_CODE); @@ -400,8 +400,6 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } - - private void writeEntries(Map map) { final int numberOfEntries = map.size(); int numberOfEntriesReadSoFar = 0; @@ -444,10 +442,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { } catch ( - Exception e4) { + Exception e4) { LOG.info("bytes position=" + bytes.position() + ",capacity=" + bytes.capacity() + ", " + - "limit=" + bytes.limit()); + "limit=" + bytes.limit()); assert bytes.limit() == bytes.capacity(); writeValue(value); @@ -469,12 +467,12 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { long remaining = bytes.remaining(); - /*LOG.info("bytes position=" + bytes.position() + ",capacity=" + bytes.capacity() + ", " + - "limit=" + bytes.limit()); -*/ + if (remaining < maxEntrySize) { + + // todo check this again as I think it may have a bug long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar); - resizeBuffer(estimatedRequiredSize * 2); + resizeBuffer(estimatedRequiredSize + maxEntrySize * 128); } } @@ -501,14 +499,14 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { assert size < Integer.MAX_VALUE; - LOG.info("bytes before position=" + bytes.position() + - ", " + - "limit=" + bytes.limit()+ ",capacity=" + bytes.capacity() ); + LOG.info("resizing to=" + size); - LOG.info("buffer before position=" + buffer.position() + + LOG.info("bytes before position=" + bytes.position() + ", " + - "limit=" + buffer.limit()+ ",capacity=" + buffer.capacity() ); + "limit=" + bytes.limit() + ",capacity=" + bytes.capacity()); + + LOG.info("buffer before"); LOG.info("resize buffer to " + size + " bytes"); final ByteBuffer result = ByteBuffer.allocateDirect((int) size).order(ByteOrder.nativeOrder()); @@ -520,7 +518,10 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { bytes = new ByteBufferBytes(result.slice()); buffer.position(0); - + buffer.limit((int) bytesPosition); + LOG.info("bytesPosition to=" + bytesPosition + ", buffer.position=" + buffer.position() + + ", " + + "limit=" + buffer.limit() + ",capacity=" + buffer.capacity()); for (int i = 0; i < bytesPosition; i++) { result.put(buffer.get()); } @@ -534,21 +535,19 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { 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() + + LOG.info("bytes after position=" + bytes.position() + ", " + - "limit=" + bytes.limit()+ ",capacity=" + bytes.capacity() ); + "limit=" + bytes.limit() + ",capacity=" + bytes.capacity()); LOG.info("buffer after position=" + buffer.position() + ", " + - "limit=" + buffer.limit()+ ",capacity=" + buffer.capacity() ); + "limit=" + buffer.limit() + ",capacity=" + buffer.capacity()); } @@ -716,7 +715,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { bytes.buffer().clear(); // the number of bytes in the response - int size = receive(2, timeoutTime).readUnsignedShort(); + int size = receive(SIZE_OF_SIZE, timeoutTime).readInt(); if (bytes.capacity() < size) bytes = new ByteBufferBytes(allocateDirect(size)); @@ -778,12 +777,13 @@ class StatelessChronicleMap implements ChronicleMap, Closeable { private void writeSizeAt(long locationOfSize) { long size = bytes.position() - locationOfSize; - bytes.writeUnsignedShort(locationOfSize, (int) size - 2); + LOG.info("writing size = " + size); + bytes.writeInt(locationOfSize, (int) size - SIZE_OF_SIZE); } private long markSizeLocation() { long position = bytes.position(); - bytes.skip(2); + bytes.skip(SIZE_OF_SIZE); return position; } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 0fe9fd7f0..096a80bc8 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -977,7 +977,7 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { in.writeByte(HEARTBEAT.ordinal()); // denotes the size in bytes - in.writeUnsignedShort(0); + in.writeInt(0); } private void writeRemoteHeartbeatInterval(long localHeartbeatInterval) { @@ -1121,17 +1121,17 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { // its set to MIN_VALUE when it should be read again if (state == NOT_SET) { - if (out.remaining() < SIZE_OF_SHORT + 1) { + if (out.remaining() < SIZE_OF_SIZE + 1) { return; } // state is used for both heartbeat and stateless state = out.readByte(); - sizeInBytes = out.readUnsignedShort(); + sizeInBytes = out.readInt(); // if the buffer is too small to read this payload we will have to grow the // size of the buffer - long requiredSize = sizeInBytes + 3; + long requiredSize = sizeInBytes + SIZE_OF_SIZE + 1; if (out.capacity() < requiredSize) { attached.entryReader.resizeBuffer(requiredSize + HEADROOM); } @@ -1727,7 +1727,7 @@ class StatelessServerConnector { private long reflectTransactionId(Bytes reader, Bytes writer) { final long sizeLocation = writer.position(); - writer.skip(3); + writer.skip(1 + AbstractChannelReplicator.SIZE_OF_SIZE); // event size + SIZE_OF_SIZE final long transactionId = reader.readLong(); writer.writeLong(transactionId); return sizeLocation; @@ -1747,8 +1747,8 @@ class StatelessServerConnector { private void writeSizeAndFlags(long locationOfSize, boolean isException, Bytes out) { final long size = out.position() - locationOfSize; - out.writeUnsignedShort(0L, (int) size); - out.writeBoolean(2L, isException); + out.writeInt(0L, (int) size); + out.writeBoolean(AbstractChannelReplicator.SIZE_OF_SIZE, isException); } private void writeException(RuntimeException e, Bytes out) { diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index e32fac525..7b70873ad 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -94,17 +94,17 @@ class UdpReplicator extends UdpChannelReplicator implements Replica.Modification final int bytesRead = in.position(); - if (bytesRead < SIZE_OF_SHORT + SIZE_OF_SHORT) + if (bytesRead < SIZE_OF_SIZE + SIZE_OF_SIZE) return; out.limit(in.position()); - final short invertedSize = out.readShort(); - final int size = out.readUnsignedShort(); + final int invertedSize = out.readInt(); + final int size = out.readInt(); // check the the first 4 bytes are the inverted len followed by the len // we do this to check that this is a valid start of entry, otherwise we throw it away - if (((short) ~size) != invertedSize) + if ( ~size != invertedSize) return; if (out.remaining() != size) @@ -157,7 +157,7 @@ class UdpReplicator extends UdpChannelReplicator implements Replica.Modification out.clear(); in.clear(); - in.skip(SIZE_OF_SHORT); + in.skip(SIZE_OF_SIZE); final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0); @@ -167,7 +167,7 @@ class UdpReplicator extends UdpChannelReplicator implements Replica.Modification } // we'll write the size inverted at the start - in.writeShort(0, ~(in.readUnsignedShort(SIZE_OF_SHORT))); + in.writeShort(0, ~(in.readUnsignedShort(SIZE_OF_SIZE))); out.limit((int) in.position()); return socketChannel.write(out); diff --git a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java index ff1a990f6..313d698bc 100644 --- a/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java +++ b/src/test/java/net/openhft/chronicle/map/StatelessClientTest.java @@ -1,6 +1,5 @@ package net.openhft.chronicle.map; -import org.junit.Ignore; import org.junit.Test; import org.testng.Assert; @@ -86,7 +85,6 @@ public class StatelessClientTest { @Test - @Ignore public void testBufferOverFlowPutAll() throws IOException, InterruptedException { final ChronicleMap serverMap; @@ -98,15 +96,12 @@ public class StatelessClientTest { statelessMap = ChronicleMapBuilder.of(Integer .class, CharSequence.class) .stateless(remoteAddress(new InetSocketAddress("localhost", 8076))).create(); - - } // server { serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class) .replicators((byte) 2, TcpReplicationConfig.of(8076)).create(); - } @@ -128,7 +123,6 @@ public class StatelessClientTest { } - @Test public void testEquals() throws IOException, InterruptedException { @@ -137,7 +131,6 @@ public class StatelessClientTest { final ChronicleMap statelessMap1; final ChronicleMap statelessMap2; - // stateless client { statelessMap1 = ChronicleMapBuilder.of(Integer @@ -146,8 +139,6 @@ public class StatelessClientTest { statelessMap2 = ChronicleMapBuilder.of(Integer .class, CharSequence.class) .stateless(remoteAddress(new InetSocketAddress("localhost", 8077))).create(); - - } // server