Merge remote-tracking branch 'origin/master'

This commit is contained in:
Rob Austin
2014-12-20 16:58:47 +00:00
24 changed files with 569 additions and 294 deletions
@@ -25,16 +25,26 @@ import java.net.InetAddress;
import java.net.NetworkInterface;
public final class UdpTransportConfig {
public static final int DEFAULT_UDP_BUFFER_SIZE = 64 * 1024;
private final InetAddress address;
private final int port;
private final @Nullable NetworkInterface networkInterface;
@Nullable
private final
NetworkInterface networkInterface;
@NotNull
private final ThrottlingConfig throttlingConfig;
private final int udpBufferSize;
private UdpTransportConfig(
InetAddress address,
int port,
@Nullable NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
ThrottlingConfig throttlingConfig,
int udpBufferSize) {
this.udpBufferSize = udpBufferSize;
if (address == null) {
throw new NullPointerException("Null address");
}
@@ -49,22 +59,23 @@ public final class UdpTransportConfig {
}
public static UdpTransportConfig multiCast(@NotNull InetAddress address, int port,
@NotNull NetworkInterface networkInterface) {
@Nullable NetworkInterface networkInterface) {
if (!address.isMulticastAddress() || networkInterface == null)
throw new IllegalArgumentException();
return create(address, port, networkInterface, ThrottlingConfig.noThrottling());
return create(address, port, networkInterface, ThrottlingConfig.noThrottling(), DEFAULT_UDP_BUFFER_SIZE);
}
public static UdpTransportConfig of(@NotNull InetAddress address, int port) {
if (address.isMulticastAddress())
throw new IllegalArgumentException();
return create(address, port, null, ThrottlingConfig.noThrottling());
return create(address, port, null, ThrottlingConfig.noThrottling(), DEFAULT_UDP_BUFFER_SIZE);
}
static UdpTransportConfig create(InetAddress address, int port,
@Nullable NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
return new UdpTransportConfig(address, port, networkInterface, throttlingConfig);
ThrottlingConfig throttlingConfig,
final int udpBufferSize1) {
return new UdpTransportConfig(address, port, networkInterface, throttlingConfig, udpBufferSize1);
}
@NotNull
@@ -129,11 +140,20 @@ public final class UdpTransportConfig {
@Nullable NetworkInterface networkInterface) {
if (!address().isMulticastAddress())
throw new IllegalArgumentException();
return create(address(), port(), networkInterface, throttlingConfig());
return create(address(), port(), networkInterface, throttlingConfig(), udpBufferSize());
}
public UdpTransportConfig throttlingConfig(@NotNull ThrottlingConfig throttlingConfig) {
ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP");
return create(address(), port(), networkInterface(), throttlingConfig);
return create(address(), port(), networkInterface(), throttlingConfig, udpBufferSize());
}
public int udpBufferSize() {
return udpBufferSize;
}
public UdpTransportConfig udpBufferSize(int udpBufferSize) {
return create(address(), port(), networkInterface(), throttlingConfig(), udpBufferSize);
}
}
@@ -410,7 +410,7 @@ abstract class AbstractChannelReplicator implements Closeable {
}
}
static class EntryCallback extends Replica.EntryCallback {
static class EntryCallback extends Replica.EntryCallback implements BufferResizer {
private final Replica.EntryExternalizable externalizable;
@@ -438,7 +438,7 @@ abstract class AbstractChannelReplicator implements Closeable {
}
private void resizeBuffer(int size) {
public Bytes resizeBuffer(int size) {
if (LOG.isDebugEnabled())
LOG.debug("resizing buffer to size=" + size);
@@ -474,7 +474,7 @@ abstract class AbstractChannelReplicator implements Closeable {
assert out.capacity() == size;
assert out.capacity() == in.capacity();
assert in.limit() == in.capacity();
return in;
}
public boolean shouldBeIgnored(final Bytes entry, final int chronicleId) {
@@ -60,11 +60,14 @@ final class ChannelProvider implements Closeable {
return channelProvider;
channelProvider = new ChannelProvider(hub);
TcpTransportAndNetworkConfig tcpConfig = hub.tcpTransportAndNetwork();
if (tcpConfig != null) {
final TcpReplicator tcpReplicator = new TcpReplicator(
channelProvider.asReplica,
channelProvider.asEntryExternalizable,
tcpConfig, null, hub.remoteNodeValidator());
tcpConfig, hub.remoteNodeValidator(), null);
channelProvider.add(tcpReplicator);
}
@@ -19,43 +19,113 @@ package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.AbstractReplication;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.chronicle.hash.replication.UdpTransportConfig;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import org.jetbrains.annotations.NotNull;
import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
final class Replicators {
static final String ONLY_UDP_WARN_MESSAGE =
"MISSING TCP REPLICATION : The UdpReplicator only attempts to read data " +
"(it does not enforce or guarantee delivery), you should use" +
"the UdpReplicator if you have a large number of nodes, and you wish" +
"to receive the data before it becomes available on TCP/IP. Since data" +
"delivery is not guaranteed, it is recommended that you only use" +
"the UDP Replicator in conjunction with a TCP Replicator";
"(it does not enforce or guarantee delivery), you should use" +
"the UdpReplicator if you have a large number of nodes, and you wish" +
"to receive the data before it becomes available on TCP/IP. Since data" +
"delivery is not guaranteed, it is recommended that you only use" +
"the UDP Replicator in conjunction with a TCP Replicator";
private Replicators() {
}
static class OutBuffer implements BufferResizer {
@NotNull
private ByteBufferBytes in;
@NotNull
private ByteBuffer out;
@NotNull
public ByteBufferBytes in() {
return in;
}
@NotNull
public ByteBuffer out() {
return out;
}
OutBuffer(final int tcpBufferSize) {
out = ByteBuffer.allocateDirect(tcpBufferSize);
in = new ByteBufferBytes(out);
}
@Override
public Bytes resizeBuffer(int newCapacity) {
if (newCapacity < out.capacity())
throw new IllegalStateException("it not possible to resize the buffer smaller");
assert newCapacity < Integer.MAX_VALUE;
final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder());
final long bytesPosition = in.position();
in = new ByteBufferBytes(result.slice());
out.position(0);
out.limit((int) bytesPosition);
int numberOfLongs = (int) bytesPosition / 8;
// chunk in longs first
for (int i = 0; i < numberOfLongs; i++) {
in.writeLong(out.getLong());
}
for (int i = numberOfLongs * 8; i < bytesPosition; i++) {
in.writeByte(out.get());
}
out = result;
assert out.capacity() == in.capacity();
assert out.capacity() == newCapacity;
assert out.capacity() == in.capacity();
assert in.limit() == in.capacity();
return in;
}
}
static Replicator tcp(final AbstractReplication replication) {
return new Replicator() {
@Override
protected Closeable applyTo(@NotNull final ChronicleMapBuilder builder,
@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable entryExternalizable,
final ChronicleMap chronicleMap)
throws IOException {
StatelessServerConnector statelessServer = new StatelessServerConnector(
builder.keyBuilder, builder.valueBuilder,
(VanillaChronicleMap) chronicleMap, builder.entrySize(true));
final ChronicleMap chronicleMap) throws IOException {
TcpTransportAndNetworkConfig tcpConfig = replication.tcpTransportAndNetwork();
assert tcpConfig != null;
TcpReplicator.StatelessClientParameters statelessClientParameters = new TcpReplicator.StatelessClientParameters(
(VanillaChronicleMap) chronicleMap,
builder.keyBuilder,
builder.valueBuilder);
return new TcpReplicator(replica, entryExternalizable,
tcpConfig,
statelessServer,
replication.remoteNodeValidator());
replication.remoteNodeValidator(), statelessClientParameters);
}
};
}
@@ -81,7 +81,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
private CloseablesManager closeables;
@NotNull
private final StatelessMapConfig config;
private int maxEntrySize;
private final Class<K> kClass;
private final Class<V> vClass;
private final boolean putReturnsNull;
@@ -139,27 +139,35 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
this.config = config;
keyReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.keyBuilder);
keyWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.keyBuilder);
valueReaderWithSize = new ReaderWithSize<>(chronicleMapBuilder.valueBuilder);
valueWriterWithSize = new WriterWithSize<>(chronicleMapBuilder.valueBuilder);
final BufferResizer outBufferResizer = new BufferResizer() {
@Override
public Bytes resizeBuffer(int newCapacity) {
return resizeBufferOutBuffer(newCapacity);
}
};
keyReaderWithSize = new ReaderWithSize(chronicleMapBuilder.keyBuilder);
keyWriterWithSize = new WriterWithSize(chronicleMapBuilder.keyBuilder, outBufferResizer);
valueReaderWithSize = new ReaderWithSize(chronicleMapBuilder.valueBuilder);
valueWriterWithSize = new WriterWithSize(chronicleMapBuilder.valueBuilder,
outBufferResizer);
this.putReturnsNull = chronicleMapBuilder.putReturnsNull();
this.removeReturnsNull = chronicleMapBuilder.removeReturnsNull();
this.maxEntrySize = chronicleMapBuilder.entrySize(true);
if (maxEntrySize < 128)
maxEntrySize = 128; // the stateless client will not work with extreemly small buffers
this.vClass = chronicleMapBuilder.valueBuilder.eClass;
this.kClass = chronicleMapBuilder.keyBuilder.eClass;
this.name = config.name();
attemptConnect(config.remoteAddress());
outBuffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder());
outBuffer = allocateDirect(128).order(ByteOrder.nativeOrder());
outBytes = new ByteBufferBytes(outBuffer.slice());
inBuffer = allocateDirect(maxEntrySize).order(ByteOrder.nativeOrder());
inBuffer = allocateDirect(128).order(ByteOrder.nativeOrder());
inBytes = new ByteBufferBytes(inBuffer.slice());
}
@@ -492,18 +500,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
return fetchObject(PUT_MAPPED, key, unaryOperator);
}
private void resizeIfRequired(int numberOfEntries, int numberOfEntriesReadSoFar, final long start) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
final long remaining = outBytes.remaining();
if (remaining < maxEntrySize) {
long estimatedRequiredSize = estimateSize(numberOfEntries, numberOfEntriesReadSoFar);
resizeBufferOutBuffer((int) (estimatedRequiredSize + maxEntrySize), start);
}
}
/**
* estimates the size based on what been completed so far
*/
@@ -515,21 +511,25 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
return (long) ((double) outBytes.position() / percentageComplete);
}
private void resizeBufferOutBuffer(int size, long start) {
private Bytes resizeBufferOutBuffer(int newCapacity) {
return resizeBufferOutBuffer(newCapacity, outBytes.position());
}
private Bytes resizeBufferOutBuffer(int newCapacity, long start) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
if (LOG.isDebugEnabled())
LOG.debug("resizing buffer to size=" + size + " ,name=" + name);
LOG.debug("resizing buffer to newCapacity=" + newCapacity + " ,name=" + name);
if (size < outBuffer.capacity())
if (newCapacity < outBuffer.capacity())
throw new IllegalStateException("it not possible to resize the buffer smaller");
assert size < Integer.MAX_VALUE;
assert newCapacity < Integer.MAX_VALUE;
final ByteBuffer result = ByteBuffer.allocate(size).order(ByteOrder.nativeOrder());
final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder());
final long bytesPosition = outBytes.position();
outBytes = new ByteBufferBytes(result.slice());
@@ -552,25 +552,27 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
assert outBuffer.capacity() == outBytes.capacity();
assert outBuffer.capacity() == size;
assert outBuffer.capacity() == newCapacity;
assert outBuffer.capacity() == outBytes.capacity();
assert outBytes.limit() == outBytes.capacity();
outBytes.position(start);
return outBytes;
}
private void resizeBufferInBuffer(int size, long start) {
private void resizeBufferInBuffer(int newCapacity, long start) {
assert !outBytesLock.isHeldByCurrentThread();
assert inBytesLock.isHeldByCurrentThread();
if (LOG.isDebugEnabled())
LOG.debug("InBuffer resizing buffer to size=" + size + " ,name=" + name);
LOG.debug("InBuffer resizing buffer to newCapacity=" + newCapacity + " ,name=" + name);
if (size < inBuffer.capacity())
if (newCapacity < inBuffer.capacity())
throw new IllegalStateException("it not possible to resize the buffer smaller");
assert size < Integer.MAX_VALUE;
assert newCapacity < Integer.MAX_VALUE;
final ByteBuffer result = ByteBuffer.allocate(size).order(ByteOrder.nativeOrder());
final ByteBuffer result = ByteBuffer.allocate(newCapacity).order(ByteOrder.nativeOrder());
final long bytesPosition = inBytes.position();
inBytes = new ByteBufferBytes(result.slice());
@@ -592,7 +594,7 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
inBuffer = result;
assert inBuffer.capacity() == inBytes.capacity();
assert inBuffer.capacity() == size;
assert inBuffer.capacity() == newCapacity;
assert inBuffer.capacity() == inBytes.capacity();
assert inBytes.limit() == inBytes.capacity();
@@ -738,10 +740,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long start = outBytes.position();
// 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, start);
final Class<?> keyClass = key.getClass();
if (!kClass.isAssignableFrom(keyClass)) {
throw new ClassCastException("key=" + key + " is of type=" + keyClass + " " +
@@ -759,10 +757,6 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
writeValueInLoop(value, valueWriter, copies);
final int len = (int) (outBytes.position() - start);
if (len > maxEntrySize)
maxEntrySize = len;
}
transactionId = send(sizeLocation, startTime);
@@ -1288,18 +1282,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
long start = outBytes.position();
for (; ; ) {
try {
return keyWriterWithSize.write(outBytes, key, copies);
} catch (IndexOutOfBoundsException e) {
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
} catch (IllegalArgumentException e) {
resizeToMessageOutBuffer(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
else
throw e;
}
return keyWriterWithSize.write(outBytes, key, copies);
}
}
@@ -1309,22 +1294,9 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
long start = outBytes.position();
for (; ; ) {
try {
return keyWriterWithSize.writeInLoop(outBytes, key, writer, copies);
} catch (IndexOutOfBoundsException e) {
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
} catch (IllegalArgumentException e) {
resizeToMessageOutBuffer(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
return keyWriterWithSize.writeInLoop(outBytes, key, writer, copies);
}
private ThreadLocalCopies writeValue(V value, ThreadLocalCopies copies) {
@@ -1332,20 +1304,13 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
assert !inBytesLock.isHeldByCurrentThread();
long start = outBytes.position();
for (; ; ) {
try {
assert outBytes.position() == start;
outBytes.limit(outBytes.capacity());
return valueWriterWithSize.write(outBytes, value, copies);
} catch (IllegalArgumentException e) {
resizeToMessageOutBuffer(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
assert outBytes.position() == start;
outBytes.limit(outBytes.capacity());
return valueWriterWithSize.write(outBytes, value, copies);
}
@SuppressWarnings("UnusedReturnValue")
@@ -1353,38 +1318,12 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
long start = outBytes.position();
for (; ; ) {
try {
assert outBytes.position() == start;
outBytes.limit(outBytes.capacity());
return valueWriterWithSize.writeInLoop(outBytes, value, writer, copies);
} catch (IllegalArgumentException e) {
resizeToMessageOutBuffer(start, e);
} catch (IllegalStateException e) {
if (e.getMessage().contains("Not enough available space"))
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
else
throw e;
}
}
}
private void resizeToMessageOutBuffer(long start, @NotNull IllegalArgumentException e) {
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
String message = e.getMessage();
if (message.startsWith(START_OF)) {
String substring = message.substring("Attempt to write ".length(), message.length());
int i = substring.indexOf(' ');
if (i != -1) {
int size = Integer.parseInt(substring.substring(0, i));
long requiresExtra = size - outBytes.remaining();
resizeBufferOutBuffer((int) (outBytes.capacity() + requiresExtra), start);
} else
resizeBufferOutBuffer((int) (outBytes.capacity() + maxEntrySize), start);
} else
throw e;
assert outBytes.position() == start;
outBytes.limit(outBytes.capacity());
return valueWriterWithSize.writeInLoop(outBytes, value, writer, copies);
}
private V readValue(long transactionId, long startTime, ThreadLocalCopies copies) {
@@ -1751,13 +1690,37 @@ class StatelessChronicleMap<K, V> implements ChronicleMap<K, V>, Closeable, Clon
if (cause instanceof IOException && cause.getMessage().contains("Not enough available space")) {
LOG.debug("resizing buffer, name=" + name);
resizeBufferOutBuffer(outBuffer.capacity() + maxEntrySize, start);
try {
resizeToMessageOutBuffer(start, e);
} catch (Exception e2) {
throw e;
}
} else
throw e;
}
}
}
private void resizeToMessageOutBuffer(long start, @NotNull Exception e) throws Exception{
assert outBytesLock.isHeldByCurrentThread();
assert !inBytesLock.isHeldByCurrentThread();
String message = e.getMessage();
if (message.startsWith("java.io.IOException: Not enough available space for writing ")) {
String substring = message.substring("java.io.IOException: Not enough available space for writing ".length(), message.length());
int i = substring.indexOf(' ');
if (i != -1) {
int size = Integer.parseInt(substring.substring(0, i));
long requiresExtra = size - outBytes.remaining();
resizeBufferOutBuffer((int) (outBytes.capacity() + requiresExtra), start);
} else
throw e;
} else
throw e;
}
@Nullable
private <R> R fetchObject(@NotNull final EventId eventId, K key, @NotNull Object object) {
@@ -62,7 +62,7 @@ interface Work {
*
* @author Rob Austin.
*/
final class TcpReplicator extends AbstractChannelReplicator implements Closeable {
final class TcpReplicator<K, V> extends AbstractChannelReplicator implements Closeable {
public static final long TIMESTAMP_FACTOR = 10000;
private static final int STATELESS_CLIENT = -127;
@@ -86,8 +86,8 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
private final Replica.EntryExternalizable externalizable;
@NotNull
private final TcpTransportAndNetworkConfig replicationConfig;
@Nullable
private final StatelessServerConnector statelessServerConnector;
private final
@Nullable
RemoteNodeValidator remoteNodeValidator;
@@ -95,19 +95,34 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
private long selectorTimeout;
StatelessClientParameters<K, V> statelessClientParameters;
static class StatelessClientParameters<K, V> {
public StatelessClientParameters(VanillaChronicleMap<K, ?, ?, V, ?, ?> map,
SerializationBuilder<K> keySerializationBuilder,
SerializationBuilder<V> valueSerializationBuilder) {
this.map = map;
this.keySerializationBuilder = keySerializationBuilder;
this.valueSerializationBuilder = valueSerializationBuilder;
}
VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
SerializationBuilder<K> keySerializationBuilder;
SerializationBuilder<V> valueSerializationBuilder;
}
/**
* @param statelessServerConnector set to NULL if not required
* @throws IOException on an io error.
*/
public TcpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable externalizable,
@NotNull final TcpTransportAndNetworkConfig replicationConfig,
@Nullable StatelessServerConnector statelessServerConnector,
@Nullable RemoteNodeValidator remoteNodeValidator) throws IOException {
@Nullable final RemoteNodeValidator remoteNodeValidator,
@Nullable final StatelessClientParameters statelessClientParameters) throws IOException {
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig()
);
this.statelessServerConnector = statelessServerConnector;
super("TcpSocketReplicator-" + replica.identifier(), replicationConfig.throttlingConfig());
this.statelessClientParameters = statelessClientParameters;
final ThrottlingConfig throttlingConfig = replicationConfig.throttlingConfig();
long throttleBucketInterval = throttlingConfig.bucketInterval(MILLISECONDS);
@@ -881,7 +896,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
/**
* Attached to the NIO selection key via methods such as {@link SelectionKey#attach(Object)}
*/
class Attached implements Replica.ModificationNotifier {
class Attached<K, V> implements Replica.ModificationNotifier {
public TcpSocketChannelEntryReader entryReader;
public TcpSocketChannelEntryWriter entryWriter;
@@ -901,16 +916,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
return handShakingComplete;
}
void clearHandShaking() {
handShakingComplete = false;
remoteIdentifier = Byte.MIN_VALUE;
remoteBootstrapTimestamp = Long.MIN_VALUE;
remoteHeartbeatInterval = heartBeatIntervalMillis;
hasRemoteHeartbeatInterval = false;
remoteModificationIterator = null;
}
/**
* called whenever there is a change to the modification iterator
*/
@@ -934,9 +939,17 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
@Nullable
public Work uncompletedWork;
private long lastSentTime;
StatelessServerConnector statelessServer;
private TcpSocketChannelEntryWriter() {
entryCallback = new EntryCallback(externalizable, replicationConfig.tcpBufferSize());
if (statelessClientParameters != null)
statelessServer = new StatelessServerConnector(
statelessClientParameters.map, entryCallback,
replicationConfig.tcpBufferSize(),
statelessClientParameters.keySerializationBuilder,
statelessClientParameters.valueSerializationBuilder);
}
public boolean isWorkIncomplete() {
@@ -960,6 +973,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
return entryCallback.in();
}
private ByteBuffer out() {
return entryCallback.out();
}
/**
* sends the identity and timestamp of this node to a remote node
*
@@ -1071,9 +1088,6 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
return len;
}
private ByteBuffer out() {
return entryCallback.out();
}
/**
* used to send an single zero byte if we have not send any data for up to the
@@ -1128,7 +1142,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
/**
* Reads map entries from a socket, this could be a client or server socket
*/
class TcpSocketChannelEntryReader {
class TcpSocketChannelEntryReader<K, V> {
public static final int HEADROOM = 1024;
public long lastHeartBeatReceived = System.currentTimeMillis();
ByteBuffer in;
@@ -1238,6 +1252,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable
boolean isStatelessClient = (state != 1);
if (isStatelessClient) {
final StatelessServerConnector statelessServerConnector = attached
.entryWriter.statelessServer;
if (statelessServerConnector == null) {
LOG.error("", new IllegalArgumentException("received an event " +
"from a stateless map, stateless maps are not " +
@@ -1338,26 +1356,33 @@ class StatelessServerConnector<K, V> {
@NotNull
private final ReaderWithSize<K> keyReaderWithSize;
@NotNull
private final WriterWithSize<K> keyWriterWithSize;
@NotNull
private final ReaderWithSize<V> valueReaderWithSize;
@NotNull
private final WriterWithSize<V> valueWriterWithSize;
@NotNull
private final VanillaChronicleMap<K, ?, ?, V, ?, ?> map;
private final double maxEntrySizeBytes;
private final int tcpBufferSize;
StatelessServerConnector(
SerializationBuilder<K> keySerializationBuilder,
SerializationBuilder<V> valueSerializationBuilder,
@NotNull VanillaChronicleMap<K, ?, ?, V, ?, ?> map, int maxEntrySizeBytes) {
@NotNull VanillaChronicleMap<K, ?, ?, V, ?, ?> map,
@NotNull final BufferResizer bufferResizer, int tcpBufferSize,
final SerializationBuilder<K> keySerializationBuilder,
final SerializationBuilder<V> valueSerializationBuilder) {
this.tcpBufferSize = tcpBufferSize;
keyReaderWithSize = new ReaderWithSize<>(keySerializationBuilder);
keyWriterWithSize = new WriterWithSize<>(keySerializationBuilder);
keyWriterWithSize = new WriterWithSize<>(keySerializationBuilder, bufferResizer);
valueReaderWithSize = new ReaderWithSize<>(valueSerializationBuilder);
valueWriterWithSize = new WriterWithSize<>(valueSerializationBuilder);
valueWriterWithSize = new WriterWithSize<>(valueSerializationBuilder, bufferResizer);
this.map = map;
this.maxEntrySizeBytes = maxEntrySizeBytes;
}
@Nullable
@@ -1784,7 +1809,7 @@ class StatelessServerConnector<K, V> {
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 (out.position() > tcpBufferSize) {
writeHeader(out, sizeLocation, count, true);
return false;
}
@@ -1826,7 +1851,7 @@ class StatelessServerConnector<K, V> {
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 (out.position() > tcpBufferSize) {
writeHeader(out, sizeLocation, count, true);
return false;
}
@@ -1859,7 +1884,7 @@ class StatelessServerConnector<K, V> {
return new Work() {
@Override
public boolean doWork(@NotNull Bytes out) {
if (out.remaining() <= maxEntrySizeBytes)
if (out.position() > tcpBufferSize)
return false;
final long sizeLocation = header(out, transactionId);
@@ -1873,7 +1898,7 @@ class StatelessServerConnector<K, V> {
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 (out.position() > tcpBufferSize) {
writeHeader(out, sizeLocation, count, true);
return false;
}
@@ -39,6 +39,9 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
private static final Logger LOG =
LoggerFactory.getLogger(UdpReplicator.class.getName());
public static final int UPD_BUFFER_SIZE = 64 * 1204;
@NotNull
private final UdpTransportConfig replicationConfig;
public UdpReplicator(@NotNull final Replica replica,
@NotNull final Replica.EntryExternalizable entryExternalizable,
@@ -47,14 +50,16 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
throws IOException {
super(replicationConfig, serializedEntrySize, replica.identifier());
this.replicationConfig = replicationConfig;
Replica.ModificationIterator modificationIterator = replica.acquireModificationIterator(
ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this);
setReader(new UdpSocketChannelEntryReader(serializedEntrySize, entryExternalizable));
setWriter(new UdpSocketChannelEntryWriter(serializedEntrySize, entryExternalizable,
modificationIterator, this));
setWriter(new UdpSocketChannelEntryWriter(replicationConfig.udpBufferSize(),
entryExternalizable,
modificationIterator, this, new Replicators.OutBuffer(UPD_BUFFER_SIZE)));
start();
}
@@ -121,17 +126,19 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
private final EntryCallback entryCallback;
private final UdpChannelReplicator udpReplicator;
private final Replicators.OutBuffer outBuffer;
private Replica.ModificationIterator modificationIterator;
UdpSocketChannelEntryWriter(final int serializedEntrySize,
UdpSocketChannelEntryWriter(final int updBufferSize,
@NotNull final Replica.EntryExternalizable externalizable,
@NotNull final Replica.ModificationIterator modificationIterator,
UdpChannelReplicator udpReplicator) {
UdpChannelReplicator udpReplicator, Replicators.OutBuffer outBuffer) {
this.udpReplicator = udpReplicator;
this.outBuffer = outBuffer;
// we make the buffer twice as large just to give ourselves headroom
entryCallback = new EntryCallback(externalizable, serializedEntrySize);
entryCallback = new EntryCallback(externalizable, updBufferSize);
this.modificationIterator = modificationIterator;
}
@@ -154,8 +161,8 @@ final class UdpReplicator extends UdpChannelReplicator implements Replica.Modifi
public int writeAll(@NotNull final DatagramChannel socketChannel)
throws InterruptedException, IOException {
final ByteBufferBytes in = entryCallback.in();
final ByteBuffer out = entryCallback.out();
final ByteBufferBytes in = outBuffer.in();
final ByteBuffer out = outBuffer.out();
out.clear();
in.clear();
@@ -25,14 +25,19 @@ import net.openhft.lang.threadlocal.Provider;
import net.openhft.lang.threadlocal.ThreadLocalCopies;
import org.jetbrains.annotations.Nullable;
import java.nio.BufferOverflowException;
final class WriterWithSize<T> {
private final SizeMarshaller sizeMarshaller;
private final Object originalWriter;
private final Provider writerProvider;
private final MetaBytesInterop originalMetaWriter;
private final MetaProvider metaWriterProvider;
private final BufferResizer bufferResizer;
WriterWithSize(SerializationBuilder<T> serializationBuilder) {
WriterWithSize(SerializationBuilder<T> serializationBuilder,
@Nullable BufferResizer bufferResizer) {
this.bufferResizer = bufferResizer;
sizeMarshaller = serializationBuilder.sizeMarshaller();
originalWriter = serializationBuilder.interop();
writerProvider = Provider.of((Class) originalWriter.getClass());
@@ -50,11 +55,25 @@ final class WriterWithSize<T> {
Object writer = writerProvider.get(copies, originalWriter);
copies = metaWriterProvider.getCopies(copies);
MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t);
sizeMarshaller.writeSize(out, metaWriter.size(writer, t));
long size = metaWriter.size(writer, t);
if (bufferResizer != null)
out = resizeIfRequired(out, size);
sizeMarshaller.writeSize(out, size);
metaWriter.write(writer, out, t);
return copies;
}
private Bytes resizeIfRequired(Bytes out, long size) {
if (out.remaining() < size + 8) {
long newCapacity = out.capacity() + (size - out.remaining()) + 8;
if (newCapacity > Integer.MAX_VALUE)
throw new BufferOverflowException();
return bufferResizer.resizeBuffer((int) newCapacity);
}
return out;
}
public ThreadLocalCopies writeNullable(Bytes out, T t, @Nullable ThreadLocalCopies copies) {
out.writeBoolean(t == null);
if (t == null)
@@ -71,7 +90,10 @@ final class WriterWithSize<T> {
@Nullable ThreadLocalCopies copies) {
copies = metaWriterProvider.getCopies(copies);
MetaBytesWriter metaWriter = metaWriterProvider.get(copies, originalMetaWriter, writer, t);
sizeMarshaller.writeSize(out, metaWriter.size(writer, t));
long size = metaWriter.size(writer, t);
if (bufferResizer != null)
out = resizeIfRequired(out, size);
sizeMarshaller.writeSize(out, size);
metaWriter.write(writer, out, t);
return copies;
}
@@ -84,3 +106,12 @@ final class WriterWithSize<T> {
return writeInLoop(out, t, writer, copies);
}
}
interface BufferResizer {
/**
* @param newCapacity
* @return the newly resize buffer
*/
public Bytes resizeBuffer(int newCapacity);
}
@@ -29,10 +29,13 @@ public class TestNesterInterfaceMarshal {
public void testReplicatedMap() throws Exception {
ChronicleMap chronicleMap = null;
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS);
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig =
TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS);
try {
ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class)
.entries(5000L).entrySize(2000)
ChronicleMapBuilder builder = ChronicleMapBuilder
.of(CharSequence.class, TestInstrumentVOInterface.class)
.entries(5000L)
.keySize(10)
.replication((byte) 1, tcpTransportAndNetworkConfig);
@@ -44,20 +47,24 @@ public class TestNesterInterfaceMarshal {
ChronicleMap<CharSequence, TestInstrumentVOInterface> map = chronicleMap;
ChronicleMap chronicleMap1 = null;
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig1 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)).heartBeatInterval(1L, TimeUnit.SECONDS);
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig1 =
TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076))
.heartBeatInterval(1L, TimeUnit.SECONDS);
try {
ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class)
ChronicleMapBuilder builder = ChronicleMapBuilder
.of(CharSequence.class, TestInstrumentVOInterface.class)
.putReturnsNull(true)
.removeReturnsNull(true)
.entries(5000L).entrySize(2000)
.entries(5000L).keySize(10)
.replication((byte) 2, tcpTransportAndNetworkConfig1);
chronicleMap1 = builder.create();
} catch (Exception e) {
System.out.println("*********************Error(s) creating launcher instrument cache: " + e);
System.out.println("*********************Error(s) creating launcher " +
"instrument cache: " + e);
}
ChronicleMap<CharSequence, TestInstrumentVOInterface> replicatedMap = chronicleMap1;
@@ -107,10 +114,13 @@ public class TestNesterInterfaceMarshal {
protected ChronicleMap createReplicatedMap1() {
ChronicleMap chronicleMap = null;
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS);
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig =
TcpTransportAndNetworkConfig.of(8076).heartBeatInterval(1L, TimeUnit.SECONDS);
try {
ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class)
.entries(5000L).entrySize(2000)
ChronicleMapBuilder builder = ChronicleMapBuilder
.of(CharSequence.class, TestInstrumentVOInterface.class)
.entries(5000L)
.keySize(10)
.replication((byte) 1, tcpTransportAndNetworkConfig);
@@ -125,20 +135,25 @@ public class TestNesterInterfaceMarshal {
protected ChronicleMap createReplicatedMap2() {
ChronicleMap chronicleMap = null;
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076)).heartBeatInterval(1L, TimeUnit.SECONDS);
TcpTransportAndNetworkConfig tcpTransportAndNetworkConfig =
TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076))
.heartBeatInterval(1L, TimeUnit.SECONDS);
try {
ChronicleMapBuilder builder = ChronicleMapBuilder.of(CharSequence.class, TestInstrumentVOInterface.class)
ChronicleMapBuilder builder = ChronicleMapBuilder
.of(CharSequence.class, TestInstrumentVOInterface.class)
.putReturnsNull(true)
.removeReturnsNull(true)
.entries(5000L).entrySize(2000)
.entries(5000L)
.keySize(10)
.replication((byte) 2, tcpTransportAndNetworkConfig);
chronicleMap = builder.create();
} catch (Exception e) {
System.out.println("*********************Error(s) creating launcher instrument cache: " + e);
System.out.println("*********************Error(s) creating launcher " +
"instrument cache: " + e);
}
return chronicleMap;
}
@@ -78,7 +78,6 @@ public class CHMLatencyTestMain {
ChronicleMap<LongValue, LongValue> countersMap =
ChronicleMapBuilder.of(LongValue.class, LongValue.class)
.entries(KEYS)
.entrySize(16)
.createPersistedTo(file);
// add keys
@@ -62,7 +62,7 @@ public class ChannelReplicationTest {
map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(2)
.entrySize(100)
.constantValueSizeBySample("EXAMPLE-1")
.instance().replicatedViaChannel(hubA.createChannel((short) 1)).create();
}
@@ -76,7 +76,7 @@ public class ChannelReplicationTest {
map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(2)
.entrySize(100)
.constantValueSizeBySample("EXAMPLE-1")
.instance().replicatedViaChannel(hubB.createChannel((short) 1)).create();
}
}
@@ -111,12 +111,12 @@ public class ChannelReplicationTest {
map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(2)
.entrySize(100)
.constantValueSizeBySample("EXAMPLE-1")
.instance().replicatedViaChannel(hubB.createChannel((short) 2)).create();
map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entries(2)
.entrySize(100)
.constantValueSizeBySample("EXAMPLE-1")
.instance().replicatedViaChannel(hubA.createChannel((short) 2)).create();
map2a.put(1, "EXAMPLE-2");
@@ -103,17 +103,12 @@ public class ChronicleMapTest {
}
private static ChronicleMap<CharSequence, LongValue> getSharedMap(
long entries, int segments, int entrySize) throws IOException {
return getSharedMap(entries, segments, entrySize, OF_4_BYTES);
}
private static ChronicleMap<CharSequence, LongValue> getSharedMap(
long entries, int segments, int entrySize, Alignment alignment)
long entries, int segments, Alignment alignment, int keySize)
throws IOException {
return ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(entries)
.minSegments(segments)
.entrySize(entrySize)
.keySize(keySize)
.entryAndValueAlignment(alignment)
.create();
}
@@ -204,7 +199,7 @@ public class ChronicleMapTest {
int count = 300;
final ChronicleMap<Object, Object> map = ChronicleMapBuilder.of(Object.class, Object.class)
.entrySize(count)
.entries(count)
.minSegments(2).create();
for (int i = 1; i < count; i++) {
@@ -414,10 +409,13 @@ public class ChronicleMapTest {
@Test
public void testAcquireWithNullContainer() throws Exception {
ChronicleMap<CharSequence, LongValue> map = getSharedMap(10 * 1000, 128, 24);
ChronicleMap<CharSequence, LongValue> map =
ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(1000)
.entryAndValueAlignment(OF_4_BYTES)
.create();
map.acquireUsing("key", new LongValue$$Native());
assertEquals(0, map.acquireUsing("key", null).getValue());
map.close();
}
@@ -454,7 +452,11 @@ public class ChronicleMapTest {
@Test
public void testGetWithNullContainer() throws Exception {
ChronicleMap<CharSequence, LongValue> map = getSharedMap(10 * 1000, 128, 24);
ChronicleMap<CharSequence, LongValue> map =
ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(10)
.entryAndValueAlignment(OF_4_BYTES)
.create();
map.acquireUsing("key", new LongValue$$Native());
assertEquals(0, map.getUsing("key", null).getValue());
@@ -463,7 +465,11 @@ public class ChronicleMapTest {
@Test
public void testGetWithoutAcquireFirst() throws Exception {
ChronicleMap<CharSequence, LongValue> map = getSharedMap(10, 1, 24);
ChronicleMap<CharSequence, LongValue> map =
ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(10)
.entryAndValueAlignment(OF_4_BYTES)
.create();
assertNull(map.getUsing("key", DataValueClasses.newDirectReference(LongValue.class)));
map.close();
@@ -485,9 +491,9 @@ public class ChronicleMapTest {
public void testAcquireAndGet() throws IOException, ClassNotFoundException,
IllegalAccessException, InstantiationException {
int entries = 3/*00 * 1000*/;
testAcquireAndGet(getSharedMap(entries, 1, 24, OF_4_BYTES), entries);
testAcquireAndGet(getSharedMap(entries, 1, 24, NO_ALIGNMENT), entries);
testAcquireAndGet(getSharedMap(entries, 1, 24, OF_8_BYTES), entries);
testAcquireAndGet(getSharedMap(entries, 1, OF_4_BYTES, 10), entries);
testAcquireAndGet(getSharedMap(entries, 1, NO_ALIGNMENT, 10), entries);
testAcquireAndGet(getSharedMap(entries, 1, OF_8_BYTES, 10), entries);
}
public void testAcquireAndGet(ChronicleMap<CharSequence, LongValue> map, int entries)
@@ -526,9 +532,9 @@ public class ChronicleMapTest {
@Test
public void testAcquireFromMultipleThreads() throws Exception {
int entries = 1000 * 1000;
testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, NO_ALIGNMENT));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, OF_4_BYTES));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, 24, OF_8_BYTES));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, NO_ALIGNMENT, 10));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, OF_4_BYTES, 10));
testAcquireFromMultipleThreads(getSharedMap(entries, 128, OF_8_BYTES, 10));
}
public void testAcquireFromMultipleThreads(ChronicleMap<CharSequence, LongValue> map)
@@ -537,7 +543,7 @@ public class ChronicleMapTest {
CharSequence key = getUserCharSequence(0);
map.acquireUsing(key, new LongValue$$Native());
int iterations = 1000;
int iterations = 10000;
int noOfThreads = 10;
CyclicBarrier barrier = new CyclicBarrier(noOfThreads);
@@ -582,7 +588,7 @@ public class ChronicleMapTest {
.entries(entries)
.actualSegments(8 * 1024)
.entryAndValueAlignment(OF_8_BYTES)
.entrySize(24);
.keySize(13);
File tmpFile = File.createTempFile("testAcquirePerf", ".deleteme");
tmpFile.deleteOnExit();
@@ -603,8 +609,10 @@ public class ChronicleMapTest {
StringBuilder sb = new StringBuilder();
long next = 50 * 1000 * 1000;
// use a factor to give up to 10 digit numbers.
int factor = Math.max(1, (int) ((10 * 1000 * 1000 * 1000L - 1) / entries));
for (long j = t % independence; j < entries + independence - 1; j += independence) {
int factor = Math.max(1,
(int) ((10 * 1000 * 1000 * 1000L - 1) / entries));
for (long j = t % independence; j < entries + independence - 1;
j += independence) {
sb.setLength(0);
sb.append("us:");
sb.append(j * factor);
@@ -625,7 +633,8 @@ public class ChronicleMapTest {
future.get();
}
long time = System.currentTimeMillis() - start;
System.out.printf("Throughput %.1f M ops/sec%n", threads * entries / independence / 1000.0 / time);
System.out.printf("Throughput %.1f M ops/sec%n",
threads * entries / independence / 1000.0 / time);
}
printStatus();
File file = map.file();
@@ -654,7 +663,7 @@ public class ChronicleMapTest {
.entries(entries)
.entryAndValueAlignment(OF_8_BYTES)
.actualSegments(256)
.entrySize(24);
.keySize(13);
File tmpFile = File.createTempFile("testAcquirePerf", ".deleteme");
tmpFile.deleteOnExit();
@@ -676,8 +685,10 @@ public class ChronicleMapTest {
long next = 50 * 1000 * 1000;
Random rand = new Random();
// use a factor to give up to 10 digit numbers.
int factor = Math.max(1, (int) ((10 * 1000 * 1000 * 1000L - 1) / entries));
for (long j = t % independence; j < entries + independence - 1; j += independence) {
int factor = Math.max(1,
(int) ((10 * 1000 * 1000 * 1000L - 1) / entries));
for (long j = t % independence; j < entries + independence - 1;
j += independence) {
sb.setLength(0);
sb.append("us:");
sb.append(j * factor);
@@ -707,7 +718,8 @@ public class ChronicleMapTest {
future.get();
}
long time = System.currentTimeMillis() - start;
System.out.printf("Throughput %.1f M ops/sec%n", threads * entries / independence / 1000.0 / time);
System.out.printf("Throughput %.1f M ops/sec%n",
threads * entries / independence / 1000.0 / time);
}
printStatus();
File file = map.file();
@@ -853,14 +865,16 @@ public class ChronicleMapTest {
}
@Test
public void testPutAndRemove() throws IOException, ClassNotFoundException, IllegalAccessException, InstantiationException {
public void testPutAndRemove() throws IOException, ClassNotFoundException,
IllegalAccessException, InstantiationException {
String TMP = System.getProperty("java.io.tmpdir");
int entries = 100 * 1000;
ChronicleMap<CharSequence, CharSequence> map =
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
.entries(entries)
.minSegments(16)
.entrySize(32)
.keySize("user:".length() + 6)
.valueSize("value:".length() + 6)
.putReturnsNull(true)
.removeReturnsNull(true).create();
StringBuilder key = new StringBuilder();
@@ -1412,7 +1426,7 @@ public class ChronicleMapTest {
final ChronicleMapBuilder<CharSequence, LongValue> builder = ChronicleMapBuilder
.of(CharSequence.class, LongValue.class)
.entries(1000)
.entrySize(16);
.keySize("x".length());
final ChronicleMap<CharSequence, LongValue> map = builder.create();
@@ -1425,7 +1439,7 @@ public class ChronicleMapTest {
ChronicleMapBuilder<CharSequence, LongValue> builder = ChronicleMapBuilder
.of(CharSequence.class, LongValue.class)
.entries(1000)
.entrySize(16);
.keySize("one".length());
final ChronicleMap<CharSequence, LongValue> map = builder.create();
@@ -1483,8 +1497,7 @@ public class ChronicleMapTest {
ChronicleMapBuilder<CharSequence, String> builder = ChronicleMapBuilder
.of(CharSequence.class, String.class)
.entries(1000)
.defaultValue("")
.entrySize(16);
.defaultValue("");
try (final ChronicleMap<CharSequence, String> map = builder.create()) {
// this will add the entry
@@ -1500,7 +1513,8 @@ public class ChronicleMapTest {
try (final ChronicleMap<CharSequence, CharSequence> map = ChronicleMapBuilder
.of(CharSequence.class, CharSequence.class)
.entries(1000)
.entrySize(40)
.keySize("one".length())
.valueSize("Hello World".length())
.defaultValue("")
.create()) {
StringBuilder value = new StringBuilder();
@@ -1518,7 +1532,7 @@ public class ChronicleMapTest {
ChronicleMapBuilder<CharSequence, LongValue> builder = ChronicleMapBuilder
.of(CharSequence.class, LongValue.class)
.entries(1000)
.entrySize(40);
.keySize("one".length());
File tmpFile = File.createTempFile("testAcquireUsingLocked", ".deleteme");
tmpFile.deleteOnExit();
final ChronicleMap<CharSequence, LongValue> map = builder.createPersistedTo(tmpFile);
@@ -34,6 +34,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
import static java.lang.Math.log10;
import static java.lang.Math.round;
import static net.openhft.chronicle.map.Alignment.OF_4_BYTES;
import static net.openhft.lang.model.DataValueClasses.newDirectReference;
import static org.junit.Assert.assertTrue;
@@ -51,12 +53,13 @@ public class EntryCountMapTest {
}
private static ChronicleMap<CharSequence, LongValue> getSharedMap(
long entries, int segments, int entrySize) throws IOException {
ChronicleMapBuilder<CharSequence, LongValue> mapBuilder = ChronicleMapBuilder.of(CharSequence
.class, LongValue.class)
long entries, int segments, int keySize) throws IOException {
ChronicleMapBuilder<CharSequence, LongValue> mapBuilder =
ChronicleMapBuilder.of(CharSequence.class, LongValue.class)
.entries(entries)
.actualSegments(segments)
.entrySize(entrySize);
.keySize(keySize)
.maxEntryOversizeFactor(1);
return mapBuilder.createPersistedTo(getPersistenceFile());
}
@@ -182,12 +185,17 @@ public class EntryCountMapTest {
System.out.printf("Score: %.2f%n", scoreCount / score);
}
private Future<Void> testEntriesMaxSize(ExecutorService es, final int segments, final int minSize, final int maxSize, final int seed) throws IOException {
private Future<Void> testEntriesMaxSize(ExecutorService es, final int segments,
final int minSize, final int maxSize, final int seed)
throws IOException {
assert minSize <= maxSize;
Random random = new Random(seed);
int counter = minSize + random.nextInt(9999 + maxSize);
final int stride = 1 + random.nextInt(100);
final ChronicleMap<CharSequence, LongValue> map = getSharedMap(minSize, segments, 32);
int maxKeySize = "key:".length() +
(int) round(log10(moreThanMaxSize(maxSize) * stride + counter)) + 1;
final ChronicleMap<CharSequence, LongValue> map =
getSharedMap(minSize, segments, maxKeySize);
return es.submit(new Callable<Void>() {
@Override
public Void call() throws Exception {
@@ -202,14 +210,17 @@ public class EntryCountMapTest {
Random random = new Random(seed);
int counter = minSize + random.nextInt(9999 + maxSize);
int stride = 1 + random.nextInt(100);
ChronicleMap<CharSequence, LongValue> map = getSharedMap(minSize, segments, 32);
int maxKeySize = "key:".length() +
(int) round(log10(moreThanMaxSize(maxSize) * stride + counter)) + 1;
ChronicleMap<CharSequence, LongValue> map = getSharedMap(minSize, segments, maxKeySize);
testEntriesMaxSize0(segments, minSize, maxSize, counter, stride, map);
}
void testEntriesMaxSize0(int segments, int minSize, int maxSize, int counter, int stride, ChronicleMap<CharSequence, LongValue> map) {
void testEntriesMaxSize0(int segments, int minSize, int maxSize, int counter, int stride,
ChronicleMap<CharSequence, LongValue> map) {
LongValue longValue = newDirectReference(LongValue.class);
try {
for (int j = 0; j < maxSize * 14 / 10 + 300; j++) {
for (int j = 0; j < moreThanMaxSize(maxSize); j++) {
String key = "key:" + counter;
counter += stride;
// give a biased hashcode distribution.
@@ -229,14 +240,20 @@ public class EntryCountMapTest {
boolean condition = minSize <= map.size() && map.size() <= maxSize;
if (!condition) {
dumpMapStats(segments, minSize, map);
assertTrue("stride: " + stride + ", seg: " + segments + ", min: " + minSize + ", size: " + map.size(), condition);
assertTrue("stride: " + stride + ", seg: " + segments + ", min: " + minSize +
", size: " + map.size(), condition);
}
} finally {
map.close();
}
}
private void dumpMapStats(int segments, int minSize, ChronicleMap<CharSequence, LongValue> map) {
private static int moreThanMaxSize(int maxSize) {
return maxSize * 14 / 10 + 300;
}
private void dumpMapStats(int segments, int minSize,
ChronicleMap<CharSequence, LongValue> map) {
long[] a = ((VanillaChronicleMap) map).segmentSizes();
System.out.println("segs: " + segments + " min: " + minSize + " was " + map.size() + " "
+ Arrays.toString(a)
@@ -50,7 +50,7 @@ public class MapByNameTest {
final MapInstanceConfig<CharSequence, CharSequence> config =
(MapInstanceConfig<CharSequence, CharSequence>)
ChronicleMapBuilder.of(CharSequence.class, CharSequence.class)
.entrySize(900)
.keySize("hello".length()).valueSize("world".length())
.entries(2)
.minSegments(2)
.replication((byte) 1, TcpTransportAndNetworkConfig.of(8244))
@@ -33,13 +33,13 @@ public class OSResizesMain {
public static void main(String[] args) throws IOException, InterruptedException {
File file = File.createTempFile("over-sized", "deleteme");
int valueSize = 1000 * 1000;
ChronicleMap<String, ByteArray> map = ChronicleMapBuilder.of(String.class, ByteArray.class)
.entrySize(50 * 1024 * 1024)
.valueSize(valueSize + 16)
.createPersistedTo(file);
byte[] chars = new byte[valueSize];
Arrays.fill(chars, (byte) '+');
ByteArray ba = new ByteArray(chars);
ChronicleMap<String, ByteArray> map = ChronicleMapBuilder.of(String.class, ByteArray.class)
.keySize("key-".length() + 4)
.constantValueSizeBySample(ba)
.createPersistedTo(file);
Arrays.fill(chars, (byte) '+');
for (int i = 0; i < 1000; i++) {
map.put("key-" + i, ba);
}
@@ -91,7 +91,7 @@ public class SerializerTest {
builder.preMapConstruction(false);
WriterWithSize valueWriterWithSize = new WriterWithSize(builder.valueBuilder);
WriterWithSize valueWriterWithSize = new WriterWithSize(builder.valueBuilder,null);
ReaderWithSize valueReaderWithSize = new ReaderWithSize(builder.valueBuilder);
valueWriterWithSize.write(out, value, null);
@@ -114,7 +114,7 @@ public class SerializerTest {
builder.preMapConstruction(false);
{
WriterWithSize keyWriterWithSize = new WriterWithSize(builder.keyBuilder);
WriterWithSize keyWriterWithSize = new WriterWithSize(builder.keyBuilder, null);
ReaderWithSize keyReaderWithSize = new ReaderWithSize(builder.keyBuilder);
keyWriterWithSize.write(out, key, null);
@@ -207,7 +207,7 @@ public class SerializerTest {
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
if (o == null || MyTestClassMarshallable.class != o.getClass()) return false;
MyTestClassMarshallable that = (MyTestClassMarshallable) o;
@@ -465,22 +465,20 @@ public class StatelessClientTest {
@Test(timeout = 10000)
public void testLargeEntries() throws IOException,
InterruptedException {
int keySize = 4;
int valueSize = 1000000;
int entrySize = keySize + valueSize;
char[] value = new char[valueSize];
Arrays.fill(value, 'X');
try (ChronicleMap<Integer, CharSequence> serverMap = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entrySize(entrySize)
String sampleValue = new String(value);
try (ChronicleMap<Integer, CharSequence> serverMap =
ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.constantValueSizeBySample(sampleValue)
.replication((byte) 2, TcpTransportAndNetworkConfig.of(8056)).create()) {
try (ChronicleMap<Integer, CharSequence> statelessMap = ChronicleMapBuilder.of(Integer
.class, CharSequence.class)
.entrySize(entrySize)
.constantValueSizeBySample(sampleValue)
.statelessClient(new InetSocketAddress("localhost", 8056)).create()) {
// for (int i = 0; i < 128; i++) {
@@ -1,6 +1,7 @@
package net.openhft.chronicle.map;
import net.openhft.chronicle.hash.replication.TcpTransportAndNetworkConfig;
import net.openhft.lang.MemoryUnit;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -24,7 +25,7 @@ public class TcpReplicationLargeEntryTest {
private ChronicleMap<Integer, CharSequence> map2;
static int s_port = 8010;
int valueSize = 1_00_000; // 1MB
int valueSize = (int) MemoryUnit.MEGABYTES.toBytes(1);
char[] valueX = new char[valueSize - 100];
char[] valueY = new char[valueSize - 100];
@@ -34,9 +35,7 @@ public class TcpReplicationLargeEntryTest {
Arrays.fill(valueX, 'X');
Arrays.fill(valueX, 'Y');
int keySize = 4;
int entrySize = keySize + valueSize;
String sampleValue = new String(valueX);
final InetSocketAddress endpoint = new InetSocketAddress("localhost", s_port + 1);
{
@@ -47,7 +46,7 @@ public class TcpReplicationLargeEntryTest {
map1 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entrySize(entrySize)
.constantValueSizeBySample(sampleValue)
.entries(2)
.actualSegments(1)
.replication((byte) 1, tcpConfig1)
@@ -62,7 +61,7 @@ public class TcpReplicationLargeEntryTest {
.tcpBufferSize(1024 * 64);
map2 = ChronicleMapBuilder.of(Integer.class, CharSequence.class)
.entrySize(entrySize)
.constantValueSizeBySample(sampleValue)
.entries(2)
.replication((byte) 2, tcpConfig2)
.instance()
@@ -80,20 +80,22 @@ public class TcpReplicationWithAcquireUsingLockedTest {
(CharSequence.class,
TestInstrumentVOInterface
.class)
.entries(5000L).entrySize(2000)
.entries(5000L).keySize("hello".length())
.replication((byte) 1, config1).create()) {
map1a = map1;
TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076));
TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig
.of(8077, new InetSocketAddress("127.0.0.1", 8076));
try (ChronicleMap<CharSequence, TestInstrumentVOInterface> map2 = ChronicleMapBuilder.of(CharSequence.class,
try (ChronicleMap<CharSequence, TestInstrumentVOInterface> map2 =
ChronicleMapBuilder.of(CharSequence.class,
TestInstrumentVOInterface.class)
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) 2, config2)
.entries(5000L).entrySize(2000).create()) {
.entries(5000L).keySize("hello".length()).create()) {
map2a = map2;
@@ -131,20 +133,22 @@ public class TcpReplicationWithAcquireUsingLockedTest {
(CharSequence.class,
TestInstrumentVOInterface
.class)
.entries(5000L).entrySize(2000)
.entries(5000L).keySize("hello".length())
.replication((byte) 1, config1).create()) {
map1a = map1;
TcpTransportAndNetworkConfig config2 = TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076));
TcpTransportAndNetworkConfig config2 =
TcpTransportAndNetworkConfig.of(8077, new InetSocketAddress("127.0.0.1", 8076));
try (ChronicleMap<CharSequence, TestInstrumentVOInterface> map2 = ChronicleMapBuilder.of(CharSequence.class,
try (ChronicleMap<CharSequence, TestInstrumentVOInterface> map2 =
ChronicleMapBuilder.of(CharSequence.class,
TestInstrumentVOInterface.class)
.putReturnsNull(true)
.removeReturnsNull(true)
.replication((byte) 2, config2)
.entries(5000L).entrySize(2000).create()) {
.entries(5000L).keySize("hello".length()).create()) {
map2a = map2;
@@ -0,0 +1,99 @@
package net.openhft.chronicle.map.example;
import net.openhft.chronicle.map.ChronicleMap;
import net.openhft.chronicle.map.ChronicleMapBuilder;
import net.openhft.lang.model.constraints.MaxSize;
import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
/**
* Created by peter on 19/12/14.
*/
public class LotsOfEntriesMain {
public static void main(String[] args)
throws IOException, ExecutionException, InterruptedException {
workEntries(true);
workEntries(false);
}
private static void workEntries(final boolean add)
throws IOException, ExecutionException, InterruptedException {
final long entries = 100_000_000;
File file = new File("/tmp/lotsOfEntries.dat");
double averageEntrySize =
1 + // key size marshalling size
// + 2 is average oversize because we append 4-letter "-key" in a loop
(Math.log(1.024) - Math.log(0.024)) * 24 + 2 + // average key size
24 + // value size
2 // average loss due to value alignment. defaults to OF_4_BYTES because
// there are floats in values
;
final ChronicleMap<CharSequence, MyFloats> map = ChronicleMapBuilder
.of(CharSequence.class, MyFloats.class)
.entries((long) (entries *
(averageEntrySize / 8.0 +
0.5))) // account average internal fragmentation --
// because we need to round up to integral number of chunks
.entrySize(8) // "chunk mode"
.createPersistedTo(file);
int threads = Runtime.getRuntime().availableProcessors();
ExecutorService es = Executors.newFixedThreadPool(threads);
long block = (entries + threads - 1) / threads;
final long start = System.nanoTime();
List<Future<?>> futures = new ArrayList<>();
for (int t = 0; t < threads; t++) {
final long startI = t * block;
final long endI = Math.min((t + 1) * block, entries);
futures.add(es.submit(new Runnable() {
@Override
public void run() {
Random rand = new Random(startI);
StringBuilder sb = new StringBuilder();
MyFloats mf = map.newValueInstance();
if (add)
for (int i = 0; i < 6; i++)
mf.setValueAt(i, i);
for (long i = startI; i < endI; i++) {
sb.setLength(0);
int length = (int) (24 / (rand.nextFloat() + 24.0 / 1000));
sb.append(i);
while (sb.length() < length)
sb.append("-key");
try {
if (add)
map.put(sb, mf);
else
map.getUsing(sb, mf);
} catch (Exception e) {
System.out.println("map.size: "+map.size());
throw e;
}
}
}
}));
}
for (Future<?> future : futures) {
future.get();
}
long time = System.nanoTime() - start;
es.shutdown();
System.out.printf("Map.size: %,d with a throughput of %.1f million/sec to %s.%n",
map.size(), entries * 1e3 / time, add ? "add" : "get");
map.close();
}
}
interface MyFloats {
public void setValueAt(@MaxSize(6) int index, float f);
public float getValueAt(int index);
}
@@ -39,9 +39,11 @@ public class ServersMapMain {
public static void startServer() throws IOException {
File file = File.createTempFile("testServersMapMain", ".deleteme");
file.deleteOnExit();
final ChronicleMap<byte[], byte[]> serverMap = ChronicleMapBuilder.of(byte[].class, byte[].class)
final ChronicleMap<byte[], byte[]> serverMap =
ChronicleMapBuilder.of(byte[].class, byte[].class)
.putReturnsNull(true)
.entrySize(50)
.constantKeySizeBySample(new byte[8])
.constantValueSizeBySample(new byte[32])
.replication((byte) 2, TcpTransportAndNetworkConfig.of(port))
.createPersistedTo(file);
System.out.println("Server started");
@@ -59,7 +61,8 @@ public class ServersMapMain {
} else {
File file = File.createTempFile("testServersMapMain", ".deleteme");
file.deleteOnExit();
TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port));
TcpTransportAndNetworkConfig tcpConfig =
TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port));
map = ChronicleMapBuilder.of(
byte[].class, byte[].class)
@@ -83,11 +86,13 @@ public class ServersMapMain {
puts += t2 - t1;
gets += t3 - t2;
if (t2 - t1 > lastAveragePut * 100 || t3 - t2 > lastAverageGet * 100)
System.out.printf("Took put/get took %.1f/%.1f us%n", (t2 - t1) / 1e3, (t3 - t2) / 1e3);
System.out.printf("Took put/get took %.1f/%.1f us%n",
(t2 - t1) / 1e3, (t3 - t2) / 1e3);
}
lastAveragePut = puts / entries;
lastAverageGet = gets / entries;
System.out.printf("Average took put/get took %.1f/%.1f us%n", lastAveragePut / 1e3, lastAverageGet / 1e3);
System.out.printf("Average took put/get took %.1f/%.1f us%n",
lastAveragePut / 1e3, lastAverageGet / 1e3);
}
}
@@ -38,10 +38,12 @@ public class StringsMapMain {
public static void startServer() throws IOException {
File file = File.createTempFile("testServersMapMain", ".deleteme");
file.deleteOnExit();
int keyAndValueSize = (int) Math.round(Math.log10(entries));
final ChronicleMap<CharSequence, CharSequence> serverMap = ChronicleMapBuilder
.of(CharSequence.class, CharSequence.class)
.putReturnsNull(true)
.entrySize(50)
.keySize(keyAndValueSize)
.valueSize(keyAndValueSize)
.replication((byte) 2, TcpTransportAndNetworkConfig.of(port))
.createPersistedTo(file);
System.out.println("Server started");
@@ -58,7 +60,8 @@ public class StringsMapMain {
} else {
File file = File.createTempFile("testServersMapMain", ".deleteme");
file.deleteOnExit();
TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port));
TcpTransportAndNetworkConfig tcpConfig =
TcpTransportAndNetworkConfig.of(port, new InetSocketAddress(hostname, port));
map = ChronicleMapBuilder
.of(CharSequence.class, CharSequence.class)
@@ -90,11 +93,13 @@ public class StringsMapMain {
puts += t2 - t1;
gets += t3 - t2;
// if (t2 - t1 > lastAveragePut * 100 || t3 - t2 > lastAverageGet * 100)
// System.out.printf("Took put/get took %.1f/%.1f us%n", (t2 - t1) / 1e3, (t3 - t2) / 1e3);
// System.out.printf("Took put/get took %.1f/%.1f us%n",
// (t2 - t1) / 1e3, (t3 - t2) / 1e3);
}
lastAveragePut = puts / entries;
lastAverageGet = gets / entries;
System.out.printf("Average took put/get took %.1f/%.1f us%n", lastAveragePut / 1e3, lastAverageGet / 1e3);
System.out.printf("Average took put/get took %.1f/%.1f us%n",
lastAveragePut / 1e3, lastAverageGet / 1e3);
}
}
@@ -50,7 +50,8 @@ public class PingPongCASLeft {
playPingPong(chm, 4, 5, true, "PingPongCASLEFT");
}
static void playPingPong(ChronicleMap<String, BondVOInterface> chm, double _coupon, double _coupon2, boolean setFirst, final String desc) {
static void playPingPong(ChronicleMap<String, BondVOInterface> chm, double _coupon,
double _coupon2, boolean setFirst, final String desc) {
BondVOInterface bond1 = newDirectReference(BondVOInterface.class);
BondVOInterface bond2 = newDirectReference(BondVOInterface.class);
BondVOInterface bond3 = newDirectReference(BondVOInterface.class);
@@ -91,6 +92,6 @@ public class PingPongCASLeft {
AffinitySupport.setThreadId();
return ChronicleMapBuilder.of(String.class, BondVOInterface.class)
.entries(16)
.entrySize(64).create();
.keySize("369604101".length()).create();
}
}
@@ -137,7 +137,7 @@ public class ProcessInstanceLimiter implements Runnable {
ChronicleMapBuilder<String, Data> builder =
ChronicleMapBuilder.of(String.class, Data.class);
builder.entries(1000);
builder.entrySize(1024);
builder.keySize((DEFAULT_PROCESS_NAME + "#").length());
this.theSharedMap = builder.createPersistedTo(new File(sharedMapPath));
Thread t = new Thread(this, "ProcessInstanceLimiter updater");
t.setDaemon(true);