Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-167 added support for larger entry sizes

Browse Source
This commit is contained in:
Rob Austin
2014-10-24 14:54:17 +01:00
parent 59da12448e
commit cbe6253ca7
6 changed files with 67 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+12 -9
View File
@@ -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;
}
src/main/java/net/openhft/chronicle/map/NodeDiscovery.java
+16 -16
View File
@@ -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
*/
src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java
+26 -26
View File
@@ -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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<? extends K, ? extends V> that = (Map<? extends K, ? extends V>) object;
@@ -311,7 +312,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
@Override
public int hashCode() {
long sizeLocation = writeEvent(HASH_CODE);
@@ -400,8 +400,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable {
}
private void writeEntries(Map<? extends K, ? extends V> map) {
final int numberOfEntries = map.size();
int numberOfEntriesReadSoFar = 0;
@@ -444,10 +442,10 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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<K, V> implements ChronicleMap<K, V>, 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;
}
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+7 -7
View File
@@ -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<K, V> {
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<K, V> {
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) {
src/main/java/net/openhft/chronicle/map/UdpReplicator.java
+6 -6
View File
@@ -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);
src/test/java/net/openhft/chronicle/map/StatelessClientTest.java
-9
View File
@@ -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<Integer, CharSequence> 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<Integer, CharSequence> statelessMap1;
final ChronicleMap<Integer, CharSequence> 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