HCOLL-134 and HCOLL-145 ( release draft cut of the code so that I can test it across my network )

This commit is contained in:
Rob Austin
2014-09-25 16:42:22 +01:00
parent 96d2fb0ee9
commit 4d889222e3
13 changed files with 693 additions and 30 deletions
@@ -136,9 +136,6 @@ abstract class AbstractChannelReplicator implements Closeable {
throttler.add(channel);
}
interface ChannelReplicatorBuilder {
}
/**
* throttles 'writes' to ensure the network is not swamped, this is achieved by periodically
* de-registering the write selector during periods of high volume.
@@ -211,9 +208,9 @@ abstract class AbstractChannelReplicator implements Closeable {
if (bytesWritten > maxBytesInInterval) {
for (SelectableChannel channel : channels) {
final SelectionKey selectionKey = channel.keyFor(selector);
if (selectionKey != null)
if (selectionKey != null) {
selectionKey.interestOps(selectionKey.interestOps() & ~OP_WRITE);
}
if (LOG.isDebugEnabled())
LOG.debug("Throttling UDP writes");
}
@@ -0,0 +1,345 @@
package net.openhft.chronicle.map;
import net.openhft.lang.collection.ATSDirectBitSet;
import net.openhft.lang.io.ByteBufferBytes;
import net.openhft.lang.io.Bytes;
import org.jetbrains.annotations.NotNull;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.util.HashSet;
import java.util.Set;
/**
* @author Rob Austin.
*/
public class NodeDiscoveryHostPortBroadcaster extends UdpChannelReplicator {
private static final byte UNUSED = (byte) -1;
static final ByteBufferBytes BOOTSTRAP_BYTES;
static {
final String BOOTSTRAP = "BOOTSTRAP";
BOOTSTRAP_BYTES = new ByteBufferBytes(ByteBuffer.allocate(2 + BOOTSTRAP.length()));
BOOTSTRAP_BYTES.write((short) BOOTSTRAP.length());
BOOTSTRAP_BYTES.append(BOOTSTRAP);
}
/**
* @param replicationConfig
* @throws java.io.IOException
*/
NodeDiscoveryHostPortBroadcaster(
@NotNull final UdpReplicationConfig replicationConfig,
final int serializedEntrySize,
RemoteNodes remoteNodes)
throws IOException {
super(replicationConfig, serializedEntrySize, UNUSED);
BytesExternalizable externalizable = new BytesExternalizableImpl(remoteNodes, this);
setReader(new UdpSocketChannelEntryReader(serializedEntrySize, externalizable));
setWriter(new UdpSocketChannelEntryWriter(serializedEntrySize, externalizable, this));
start();
}
private static class UdpSocketChannelEntryReader implements EntryReader {
private final ByteBuffer in;
private final ByteBufferBytes out;
private final BytesExternalizable externalizable;
/**
* @param serializedEntrySize the maximum size of an entry include the meta data
* @param externalizable supports reading and writing serialize entries
*/
UdpSocketChannelEntryReader(final int serializedEntrySize,
@NotNull final BytesExternalizable externalizable) {
// we make the buffer twice as large just to give ourselves headroom
in = ByteBuffer.allocateDirect(serializedEntrySize * 2);
out = new ByteBufferBytes(in);
out.limit(0);
in.clear();
this.externalizable = externalizable;
}
/**
* reads entries from the socket till it is empty
*
* @param socketChannel the socketChannel that we will read from
* @throws IOException
* @throws InterruptedException
*/
public void readAll(@NotNull final DatagramChannel socketChannel) throws IOException,
InterruptedException {
out.clear();
in.clear();
socketChannel.receive(in);
final int bytesRead = in.position();
if (bytesRead < SIZE_OF_SHORT + SIZE_OF_SHORT)
return;
out.limit(in.position());
final short invertedSize = out.readShort();
final int size = out.readUnsignedShort();
// 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)
return;
if (out.remaining() != size)
return;
externalizable.readExternalBytes(out);
}
}
private static class UdpSocketChannelEntryWriter implements EntryWriter {
private final ByteBuffer out;
private final ByteBufferBytes in;
private final BytesExternalizable externalizable;
private UdpChannelReplicator udpReplicator;
UdpSocketChannelEntryWriter(final int serializedEntrySize,
@NotNull final BytesExternalizable externalizable,
@NotNull final UdpChannelReplicator udpReplicator) {
this.udpReplicator = udpReplicator;
// we make the buffer twice as large just to give ourselves headroom
out = ByteBuffer.allocateDirect(serializedEntrySize * 2);
in = new ByteBufferBytes(out);
// write bootstrap to the 'in' buffer
{
// skip 2 bytes where we write the size
in.skip(SIZE_OF_SHORT);
in.write(BOOTSTRAP_BYTES);
// now write the size at the start
in.writeShort(0, (int) in.position() - SIZE_OF_SHORT);
}
this.externalizable = externalizable;
}
/**
* writes all the entries that have changed, to the tcp socket
*
* @param socketChannel
* @param modificationIterator
* @throws InterruptedException
* @throws java.io.IOException
*/
/**
* update that are throttled are rejected.
*
* @param socketChannel the socketChannel that we will write to
* @throws InterruptedException
* @throws IOException
*/
public int writeAll(@NotNull final DatagramChannel socketChannel)
throws InterruptedException, IOException {
out.clear();
in.clear();
in.skip(SIZE_OF_SHORT);
// we'll write the size inverted at the start
in.writeShort(0, ~(in.readUnsignedShort(SIZE_OF_SHORT)));
out.limit((int) in.position());
int start = out.position();
// writes the contents of
externalizable.writeExternalBytes(in);
udpReplicator.disableWrites();
// number of bytes written
return out.position() - start;
}
}
}
/**
* supports reading and writing serialize entries
*/
interface BytesExternalizable {
/**
* The map implements this method to save its contents.
*
* @param destination a buffer the entry will be written to, the segment may reject this operation and add
* zeroBytes, if the identifier in the entry did not match the maps local
*/
void writeExternalBytes(@NotNull Bytes destination);
/**
* The map implements this method to restore its contents. This method must read the values in the same
* sequence and with the same types as were written by {@code writeExternalEntry()}. This method is
* typically called when we receive a remote replication event, this event could originate from either a
* remote {@code put(K key, V value)} or {@code remove(Object key)}
*
* @param source bytes to read an entry from
*/
void readExternalBytes(@NotNull Bytes source);
}
class RemoteNodes {
private final Bytes activeIdentifiersBitSetBytes;
private Set<InetSocketAddress> inetSocketAddresses;
private final ATSDirectBitSet atsDirectBitSet;
/**
* @param inetSocketAddresses the host and ports which we should connect to for TCP replication
* @param activeIdentifiersBitSetBytes byte sof a bitset containing the known identifiers
*/
RemoteNodes(final Bytes activeIdentifiersBitSetBytes) {
this.inetSocketAddresses = new HashSet<InetSocketAddress>();
this.activeIdentifiersBitSetBytes = activeIdentifiersBitSetBytes;
this.atsDirectBitSet = new ATSDirectBitSet(this.activeIdentifiersBitSetBytes);
}
public Set<InetSocketAddress> inetSocketAddresses() {
return inetSocketAddresses;
}
public Bytes activeIdentifierBytes() {
return activeIdentifiersBitSetBytes;
}
public void add(InetSocketAddress inetSocketAddress) {
inetSocketAddresses.add(inetSocketAddress);
}
public ATSDirectBitSet activeIdentifierBitSet() {
return atsDirectBitSet;
}
}
class BytesExternalizableImpl implements BytesExternalizable {
private final RemoteNodes allNodes;
private final Replica.ModificationNotifier modificationNotifier;
//private final InetSocketAddress localInetSocketAddress;
public BytesExternalizableImpl(final RemoteNodes allNodes,
final Replica.ModificationNotifier modificationNotifier) {
this.allNodes = allNodes;
this.modificationNotifier = modificationNotifier;
// todo this should only be added once we have connected - we will add our host and port to allNodes
//this.localInetSocketAddress = new InetSocketAddress(localHost, localPort);
//this.allNodes.add(localInetSocketAddress);
}
/**
* this is used to tell nodes that are connecting to us which host and ports are in our grid, along with
* all the identifiers.
*
* @param destination a buffer the entry will be written to, the segment may reject this operation and add
* zeroBytes, if the identifier in the entry did not match the maps local
*/
@Override
public void writeExternalBytes(@NotNull Bytes destination) {
final Set<InetSocketAddress> inetSocketAddresses = allNodes.inetSocketAddresses();
// write the number of hosts and ports
destination.write((short) inetSocketAddresses.size());
// write all the host and ports
for (InetSocketAddress inetSocketAddress : inetSocketAddresses) {
destination.writeUTF(inetSocketAddress.getHostName());
destination.writeInt(inetSocketAddress.getPort());
}
destination.write(allNodes.activeIdentifierBytes());
}
@Override
public void readExternalBytes(@NotNull Bytes source) {
if (source.equals(NodeDiscoveryHostPortBroadcaster.BOOTSTRAP_BYTES)) {
// used to wake up the OP_WRITE
modificationNotifier.onChange();
return;
}
// the number of host and ports
int count = source.readShort();
for (int i = 0; i < count; i++) {
final String host = source.readUTF();
final int port = source.readInt();
allNodes.add(new InetSocketAddress(host, port));
}
int identifierBitSetBytes = source.readShort();
source.limit(source.position() + identifierBitSetBytes);
final ATSDirectBitSet sourceBitSet = new ATSDirectBitSet(source);
final ATSDirectBitSet resultBitSet = allNodes.activeIdentifierBitSet();
// merges the source into the destination and returns the destination
orBitSets(sourceBitSet, resultBitSet);
}
/**
* merges the source into the destination and returns the destination
*
* @param source
* @param destination
* @return
*/
private ATSDirectBitSet orBitSets(ATSDirectBitSet source, ATSDirectBitSet destination) {
// merge the two bit-sets together via 'or'
long numberOfLongs = source.size() / 8;
for (int longIndex = 1; longIndex < numberOfLongs; longIndex++) {
destination.or(longIndex, source.getLong(longIndex));
}
return destination;
}
}
@@ -223,7 +223,7 @@ public final class ReplicatingCluster implements Closeable {
final int chronicleChannel = bytes.readUnsignedShort();
final long lastModificationTime = bytes.readLong();
// this could be null if once node has a chronicle channel before the other
// this could be null if one node has a chronicle channel before the other
if (chronicleChannels[chronicleChannel] != null) {
chronicleChannels[chronicleChannel].acquireModificationIterator(remoteIdentifier, NOP)
.dirtyEntries(lastModificationTime);
@@ -315,6 +315,9 @@ public final class ReplicatingCluster implements Closeable {
final PayloadProvider iterator = new PayloadProvider() {
// todo it maybe possibel to replace this with a single atomic refernce,
// todo this way no aditional objects will be created
final Queue<Bytes> payloads = new LinkedTransferQueue<Bytes>();
@Override
@@ -16,8 +16,6 @@
package net.openhft.chronicle.map;
import com.google.auto.value.AutoValue;
import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.HashSet;
@@ -28,7 +26,7 @@ import static java.util.Arrays.asList;
import static java.util.Collections.unmodifiableSet;
import static java.util.concurrent.TimeUnit.SECONDS;
@AutoValue
public abstract class TcpReplicationConfig {
private static final int DEFAULT_PACKET_SIZE = 1024 * 8;
@@ -65,7 +63,7 @@ public abstract class TcpReplicationConfig {
static TcpReplicationConfig create(int serverPort, Set<InetSocketAddress> endpoints,
int packetSize, ThrottlingConfig throttlingConfig,
long heartBeatInterval, TimeUnit heartBeatIntervalUnit) {
return new AutoValue_TcpReplicationConfig(serverPort, endpoints, packetSize,
return new TcpReplicationConfigBean(serverPort, endpoints, packetSize,
throttlingConfig, heartBeatInterval, heartBeatIntervalUnit);
}
@@ -0,0 +1,116 @@
package net.openhft.chronicle.map;
import java.net.InetSocketAddress;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@javax.annotation.Generated("com.google.auto.value.processor.AutoValueProcessor")
final class TcpReplicationConfigBean extends TcpReplicationConfig {
private final int serverPort;
private final Set<InetSocketAddress> endpoints;
private final int packetSize;
private final ThrottlingConfig throttlingConfig;
private final long heartBeatInterval;
private final TimeUnit heartBeatIntervalUnit;
TcpReplicationConfigBean(
int serverPort,
Set<InetSocketAddress> endpoints,
int packetSize,
ThrottlingConfig throttlingConfig,
long heartBeatInterval,
TimeUnit heartBeatIntervalUnit) {
this.serverPort = serverPort;
if (endpoints == null) {
throw new NullPointerException("Null endpoints");
}
this.endpoints = endpoints;
this.packetSize = packetSize;
if (throttlingConfig == null) {
throw new NullPointerException("Null throttlingConfig");
}
this.throttlingConfig = throttlingConfig;
this.heartBeatInterval = heartBeatInterval;
if (heartBeatIntervalUnit == null) {
throw new NullPointerException("Null heartBeatIntervalUnit");
}
this.heartBeatIntervalUnit = heartBeatIntervalUnit;
}
@Override
public int serverPort() {
return serverPort;
}
@Override
public Set<InetSocketAddress> endpoints() {
return endpoints;
}
@Override
public int packetSize() {
return packetSize;
}
@Override
public ThrottlingConfig throttlingConfig() {
return throttlingConfig;
}
@Override
long heartBeatInterval() {
return heartBeatInterval;
}
@Override
TimeUnit heartBeatIntervalUnit() {
return heartBeatIntervalUnit;
}
@Override
public String toString() {
return "TcpReplicationConfig{"
+ "serverPort=" + serverPort
+ ", endpoints=" + endpoints
+ ", packetSize=" + packetSize
+ ", throttlingConfig=" + throttlingConfig
+ ", heartBeatInterval=" + heartBeatInterval
+ ", heartBeatIntervalUnit=" + heartBeatIntervalUnit
+ "}";
}
@Override
public boolean equals(Object o) {
if (o == this) {
return true;
}
if (o instanceof TcpReplicationConfig) {
TcpReplicationConfig that = (TcpReplicationConfig) o;
return (this.serverPort == that.serverPort())
&& (this.endpoints.equals(that.endpoints()))
&& (this.packetSize == that.packetSize())
&& (this.throttlingConfig.equals(that.throttlingConfig()))
&& (this.heartBeatInterval == that.heartBeatInterval())
&& (this.heartBeatIntervalUnit.equals(that.heartBeatIntervalUnit()));
}
return false;
}
@Override
public int hashCode() {
int h = 1;
h *= 1000003;
h ^= serverPort;
h *= 1000003;
h ^= endpoints.hashCode();
h *= 1000003;
h ^= packetSize;
h *= 1000003;
h ^= throttlingConfig.hashCode();
h *= 1000003;
h ^= (heartBeatInterval >>> 32) ^ heartBeatInterval;
h *= 1000003;
h ^= heartBeatIntervalUnit.hashCode();
return h;
}
}
@@ -16,13 +16,12 @@
package net.openhft.chronicle.map;
import com.google.auto.value.AutoValue;
import java.util.concurrent.TimeUnit;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
@AutoValue
public abstract class ThrottlingConfig {
private static final long DEFAULT_BUCKET_INTERVAL = 100L;
@@ -48,8 +47,8 @@ public abstract class ThrottlingConfig {
}
/**
* Returns a config which throttle the specified max bits per the given time unit
* with default bucketing interval.
* Returns a config which throttle the specified max bits per the given time unit with default bucketing
* interval.
*
* @param maxBits the preferred maximum bits
* @param perUnit the time unit per which maximum bits specified
@@ -73,7 +72,7 @@ public abstract class ThrottlingConfig {
static ThrottlingConfig create(
long throttle, TimeUnit throttlePerUnit,
long throttleBucketInterval, TimeUnit throttleBucketIntervalUnit) {
return new AutoValue_ThrottlingConfig(throttle, throttlePerUnit,
return new ThrottlingConfigBean(throttle, throttlePerUnit,
throttleBucketInterval, throttleBucketIntervalUnit);
}
@@ -82,8 +81,8 @@ public abstract class ThrottlingConfig {
abstract TimeUnit throttlePerUnit();
/**
* Returns maximum bits per the given time unit, i. e. the throttling. {@code 0} (zero)
* designates there is no throttling.
* Returns maximum bits per the given time unit, i. e. the throttling. {@code 0} (zero) designates there
* is no throttling.
*
* @param perUnit maximum bits is returned per this time unit
* @return maximum bits per the given time unit, or {@code 0} if there is no throttling
@@ -97,9 +96,8 @@ public abstract class ThrottlingConfig {
abstract TimeUnit bucketIntervalUnit();
/**
* Returns the throttle bucketing interval in the given time units.
* <p/>
* <p>Default throttle bucketing interval is 100 millis.
* Returns the throttle bucketing interval in the given time units. <p/> <p>Default throttle bucketing
* interval is 100 millis.
*
* @param unit the time unit of the interval
* @return the bucketing interval for throttling in the given time units
@@ -0,0 +1,87 @@
package net.openhft.chronicle.map;
import java.util.concurrent.TimeUnit;
@javax.annotation.Generated("com.google.auto.value.processor.AutoValueProcessor")
final class ThrottlingConfigBean extends ThrottlingConfig {
private final long throttle;
private final TimeUnit throttlePerUnit;
private final long bucketInterval;
private final TimeUnit bucketIntervalUnit;
ThrottlingConfigBean(
long throttle,
TimeUnit throttlePerUnit,
long bucketInterval,
TimeUnit bucketIntervalUnit) {
this.throttle = throttle;
if (throttlePerUnit == null) {
throw new NullPointerException("Null throttlePerUnit");
}
this.throttlePerUnit = throttlePerUnit;
this.bucketInterval = bucketInterval;
if (bucketIntervalUnit == null) {
throw new NullPointerException("Null bucketIntervalUnit");
}
this.bucketIntervalUnit = bucketIntervalUnit;
}
@Override
long throttle() {
return throttle;
}
@Override
TimeUnit throttlePerUnit() {
return throttlePerUnit;
}
@Override
long bucketInterval() {
return bucketInterval;
}
@Override
TimeUnit bucketIntervalUnit() {
return bucketIntervalUnit;
}
@Override
public String toString() {
return "ThrottlingConfig{"
+ "throttle=" + throttle
+ ", throttlePerUnit=" + throttlePerUnit
+ ", bucketInterval=" + bucketInterval
+ ", bucketIntervalUnit=" + bucketIntervalUnit
+ "}";
}
@Override
public boolean equals(Object o) {
if (o == this) {
return true;
}
if (o instanceof ThrottlingConfig) {
ThrottlingConfig that = (ThrottlingConfig) o;
return (this.throttle == that.throttle())
&& (this.throttlePerUnit.equals(that.throttlePerUnit()))
&& (this.bucketInterval == that.bucketInterval())
&& (this.bucketIntervalUnit.equals(that.bucketIntervalUnit()));
}
return false;
}
@Override
public int hashCode() {
int h = 1;
h *= 1000003;
h ^= (throttle >>> 32) ^ throttle;
h *= 1000003;
h ^= throttlePerUnit.hashCode();
h *= 1000003;
h ^= (bucketInterval >>> 32) ^ bucketInterval;
h *= 1000003;
h ^= bucketIntervalUnit.hashCode();
return h;
}
}
@@ -209,6 +209,8 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.
final SelectionKey selectionKey = writeChannel.keyFor(this.selector);
if (selectionKey != null)
selectionKey.interestOps(selectionKey.interestOps() | OP_WRITE);
// we have just enabled it, so don't have to do this again.
shouldEnableOpWrite = false;
} catch (Exception e) {
LOG.error("", e);
}
@@ -16,14 +16,13 @@
package net.openhft.chronicle.map;
import com.google.auto.value.AutoValue;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import java.net.InetAddress;
import java.net.NetworkInterface;
@AutoValue
public abstract class UdpReplicationConfig {
/**
@@ -36,6 +35,9 @@ public abstract class UdpReplicationConfig {
public static UdpReplicationConfig simple(@NotNull InetAddress address, int port) {
if (address.isMulticastAddress())
throw new IllegalArgumentException();
return create(address, port, null, ThrottlingConfig.noThrottling());
}
@@ -49,7 +51,7 @@ public abstract class UdpReplicationConfig {
static UdpReplicationConfig create(InetAddress address, int port,
NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
return new AutoValue_UdpReplicationConfig(address, port, networkInterface,
return new UdpReplicationConfigBean(address, port, networkInterface,
throttlingConfig);
}
@@ -0,0 +1,89 @@
package net.openhft.chronicle.map;
import java.net.InetAddress;
import java.net.NetworkInterface;
final class UdpReplicationConfigBean extends UdpReplicationConfig {
private final InetAddress address;
private final int port;
private final NetworkInterface networkInterface;
private final ThrottlingConfig throttlingConfig;
UdpReplicationConfigBean(
InetAddress address,
int port,
NetworkInterface networkInterface,
ThrottlingConfig throttlingConfig) {
if (address == null) {
throw new NullPointerException("Null address");
}
this.address = address;
this.port = port;
this.networkInterface = networkInterface;
if (throttlingConfig == null) {
throw new NullPointerException("Null throttlingConfig");
}
this.throttlingConfig = throttlingConfig;
}
@Override
public InetAddress address() {
return address;
}
@Override
public int port() {
return port;
}
@Override
public NetworkInterface networkInterface() {
return networkInterface;
}
@Override
public ThrottlingConfig throttlingConfig() {
return throttlingConfig;
}
@Override
public String toString() {
return "UdpReplicationConfig{"
+ "address=" + address
+ ", port=" + port
+ ", networkInterface=" + networkInterface
+ ", throttlingConfig=" + throttlingConfig
+ "}";
}
@Override
public boolean equals(Object o) {
if (o == this) {
return true;
}
if (o instanceof UdpReplicationConfig) {
UdpReplicationConfig that = (UdpReplicationConfig) o;
return (this.address.equals(that.address()))
&& (this.port == that.port())
&& (this.networkInterface.equals(that.networkInterface()))
&& (this.throttlingConfig.equals(that.throttlingConfig()));
}
return false;
}
@Override
public int hashCode() {
int h = 1;
h *= 1000003;
h ^= address.hashCode();
h *= 1000003;
h ^= port;
h *= 1000003;
h ^= networkInterface.hashCode();
h *= 1000003;
h ^= throttlingConfig.hashCode();
return h;
}
}