diff --git a/src/main/java/net/openhft/chronicle/hash/ChannelProvider.java b/src/main/java/net/openhft/chronicle/hash/ChannelProvider.java new file mode 100644 index 000000000..54a621cf2 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/ChannelProvider.java @@ -0,0 +1,436 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash; + + +import net.openhft.chronicle.map.*; +import net.openhft.lang.collection.DirectBitSet; +import net.openhft.lang.collection.SingleThreadedDirectBitSet; +import net.openhft.lang.io.ByteBufferBytes; +import net.openhft.lang.io.Bytes; +import org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArraySet; +import java.util.concurrent.LinkedTransferQueue; +import java.util.concurrent.atomic.AtomicReferenceArray; + +import static java.lang.Math.min; +import static java.nio.ByteBuffer.wrap; +import static net.openhft.chronicle.map.Replica.EntryExternalizable; +import static net.openhft.chronicle.map.Replica.ModificationIterator; +import static net.openhft.chronicle.map.Replica.ModificationNotifier.NOP; + +/** + * @author Rob Austin. + */ +public final class ChannelProvider implements Closeable { + private static final Logger LOG = LoggerFactory.getLogger(ChannelProvider.class.getName()); + + private static final byte BOOTSTRAP_MESSAGE = 'B'; + final EntryExternalizable asEntryExternalizable = new EntryExternalizable() { + /** + * writes the entry to the chronicle channel provided + * + * @param entry the byte location of the entry to be stored + * @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 + * @param chronicleChannel used in cluster into identify the canonical map or queue + */ + @Override + public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, + int chronicleChannel) { + destination.writeStopBit(chronicleChannel); + channelEntryExternalizables[chronicleChannel] + .writeExternalEntry(entry, destination, chronicleChannel); + } + + @Override + public void readExternalEntry(@NotNull Bytes source) { + final int chronicleId = (int) source.readStopBit(); + if (chronicleId < chronicleChannels.length) + channelEntryExternalizables[chronicleId].readExternalEntry(source); + else + LOG.info("skipped entry with chronicleId=" + chronicleId + ", "); + } + }; + private final byte localIdentifier; + final Replica asReplica = new Replica() { + + @Override + public byte identifier() { + return localIdentifier; + } + + @Override + public ModificationIterator acquireModificationIterator( + final short remoteIdentifier, final ModificationNotifier notifier) { + final ModificationIterator result = modificationIterator.get(remoteIdentifier); + if (result != null) + return result; + + final ModificationIterator result0 = new ModificationIterator() { + + @Override + public boolean hasNext() { + // old-style iteration to avoid 1) iterator object creation + // 2) ConcurrentModificationException + for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { + final ModificationIterator modificationIterator = + chronicleChannelList.get(i) + .acquireModificationIterator(remoteIdentifier, notifier); + if (modificationIterator.hasNext()) + return true; + } + return false; + } + + @Override + public boolean nextEntry(@NotNull EntryCallback callback, + final int na) { + for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { + Replica chronicleChannel = chronicleChannelList.get(i); + final ModificationIterator modificationIterator = chronicleChannel + .acquireModificationIterator(remoteIdentifier, notifier); + if (modificationIterator.nextEntry(callback, chronicleChannelIds.get(i))) + return true; + } + return false; + } + + @Override + public void dirtyEntries(long fromTimeStamp) { + for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { + chronicleChannelList.get(i) + .acquireModificationIterator(remoteIdentifier, notifier) + .dirtyEntries(fromTimeStamp); + notifier.onChange(); + } + } + }; + + modificationIterator.set((int) remoteIdentifier, result0); + return result0; + + } + + /** + * gets the earliest modification time for all of the chronicles + */ + @Override + public long lastModificationTime(byte remoteIdentifier) { + long t = 0; + // not including the SystemQueue at index 0 + for (int i = 1, len = chronicleChannelList.size(); i < len; i++) { + t = (t == 0) ? chronicleChannels[i].lastModificationTime(remoteIdentifier) : + min(t, chronicleChannels[i].lastModificationTime(remoteIdentifier)); + } + return t; + } + + @Override + public void close() throws IOException { + ChannelProvider.this.close(); + } + }; + private final int maxEntrySize; + private final Replica[] chronicleChannels; + private final List chronicleChannelList; + private final List chronicleChannelIds; + private final EntryExternalizable[] channelEntryExternalizables; + private final AtomicReferenceArray systemModificationIterator = + new AtomicReferenceArray(128); + private final DirectBitSet systemModificationIteratorBitSet = + newBitSet(systemModificationIterator.length()); + private final AtomicReferenceArray modificationIterator = + new AtomicReferenceArray(128); + private final Set replicators = new CopyOnWriteArraySet(); + + ChannelProvider(ChannelProviderBuilder builder) { + localIdentifier = builder.identifier; + maxEntrySize = builder.maxEntrySize; + chronicleChannels = new Replica[builder.maxNumberOfChronicles]; + channelEntryExternalizables = new EntryExternalizable[builder.maxNumberOfChronicles]; + chronicleChannelList = new ArrayList(); + chronicleChannelIds = new ArrayList(); + MessageHandler systemMessageHandler = new MessageHandler() { + @Override + public void onMessage(Bytes bytes) { + final byte type = bytes.readByte(); + if (type == BOOTSTRAP_MESSAGE) { + onBootstrapMessage(bytes); + } else { + LOG.info("message of type=" + type + " was ignored."); + } + } + }; + SystemQueue systemMessageQueue = new SystemQueue( + systemModificationIteratorBitSet, systemModificationIterator, systemMessageHandler); + add((short) 0, systemMessageQueue.asReplica, systemMessageQueue.asEntryExternalizable); + } + + public int maxEntrySize() { + return maxEntrySize; + } + + /** + * creates a bit set based on a number of bits + * + * @param numberOfBits the number of bits the bit set should include + * @return a new DirectBitSet backed by a byteBuffer + */ + private static DirectBitSet newBitSet(int numberOfBits) { + final ByteBufferBytes bytes = new ByteBufferBytes(wrap(new byte[(numberOfBits + 7) / 8])); + return new SingleThreadedDirectBitSet(bytes); + } + + /** + * Returns a replicator, dedicated to the specified channel. Channel is basically just a number, that + * should correspond on different servers for instances of the same replicated map. Only one replicator + * per channel could be obtained from a single {@code ReplicatingChannel}. The returned replicator could + * be applied to a map at most once. + * + * @param channel number to create + * @return a replicator, dedicated to the specified channel + */ + public ChronicleChannel createChannel(short channel) { + return new ChronicleChannel(channel); + } + + /** + * called whenever we receive a bootstrap message + */ + private void onBootstrapMessage(Bytes bytes) { + final short remoteIdentifier = bytes.readByte(); + final int chronicleChannel = bytes.readUnsignedShort(); + final long lastModificationTime = bytes.readLong(); + + // 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); + } + } + + private ByteBufferBytes toBootstrapMessage(short chronicleChannel, final long lastModificationTime) { + final ByteBufferBytes writeBuffer = new ByteBufferBytes(ByteBuffer.allocate(1 + 1 + 2 + 8)); + writeBuffer.writeByte(BOOTSTRAP_MESSAGE); + writeBuffer.writeByte(localIdentifier); + writeBuffer.writeUnsignedShort(chronicleChannel); + writeBuffer.writeLong(lastModificationTime); + writeBuffer.flip(); + return writeBuffer; + } + + private void add(short chronicleChannel, + Replica replica, EntryExternalizable entryExternalizable) { + if (chronicleChannels[chronicleChannel] != null) { + throw new IllegalStateException("chronicleId=" + chronicleChannel + + " is already in use."); + } + chronicleChannels[chronicleChannel] = replica; + chronicleChannelList.add(replica); + chronicleChannelIds.add((int) chronicleChannel); + channelEntryExternalizables[chronicleChannel] = entryExternalizable; + + if (chronicleChannel == 0) + return; + + for (int i = (int) systemModificationIteratorBitSet.nextSetBit(0); i > 0; + i = (int) systemModificationIteratorBitSet.nextSetBit(i + 1)) { + byte remoteIdentifier = (byte) i; + final long lastModificationTime = replica.lastModificationTime(remoteIdentifier); + final ByteBufferBytes message = + toBootstrapMessage(chronicleChannel, lastModificationTime); + systemModificationIterator.get(remoteIdentifier).addPayload(message); + } + + } + + @Override + public void close() throws IOException { + for (Closeable replicator : replicators) { + replicator.close(); + } + // indexed reverse loop to avoid ConcurrentModificationException and remove elements from + // the end of the loop for correct iteration + for (int i = chronicleChannelList.size() - 1; i >= 0; i--) { + Replica chronicleChannel = chronicleChannelList.get(i); + chronicleChannel.close(); + } + } + + void add(Closeable replicator) { + replicators.add(replicator); + } + + private interface MessageHandler { + void onMessage(Bytes bytes); + } + + private interface PayloadProvider extends ModificationIterator { + void addPayload(final Bytes bytes); + } + + /** + * used to send system messages such as bootstrap from one remote node to another, it also can be used in + * a broadcast context + */ + static class SystemQueue { + + + final Replica asReplica = new Replica() { + + @Override + public byte identifier() { + return 0; + } + + @Override + public ModificationIterator acquireModificationIterator( + final short remoteIdentifier, final ModificationNotifier modificationNotifier) { + + final ModificationIterator result = systemModificationIterator.get(remoteIdentifier); + + if (result != null) + return result; + + 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 payloads = new LinkedTransferQueue(); + + @Override + public boolean hasNext() { + return payloads.peek() != null; + } + + @Override + public boolean nextEntry(@NotNull EntryCallback callback, int na) { + final Bytes bytes = payloads.poll(); + if (bytes == null) + return false; + callback.onEntry(bytes, 0); + return true; + } + + @Override + public void dirtyEntries(long fromTimeStamp) { + // do nothing + } + + @Override + public void addPayload(Bytes bytes) { + if (bytes.remaining() == 0) + return; + payloads.add(bytes); + // notifies that a change has been made, this will nudge the OP_WRITE + // selector to push this update out over the nio socket + modificationNotifier.onChange(); + } + }; + + systemModificationIterator.set(remoteIdentifier, iterator); + systemModificationIteratorBitSet.set(remoteIdentifier); + + return iterator; + + } + + @Override + public long lastModificationTime(byte remoteIdentifier) { + return 0; + } + + @Override + public void close() throws IOException { + // do nothing + } + }; + final EntryExternalizable asEntryExternalizable = new EntryExternalizable() { + @Override + public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination, + int na) { + destination.write(entry); + } + + @Override + public void readExternalEntry(@NotNull Bytes source) { + messageHandler.onMessage(source); + } + }; + private final DirectBitSet systemModificationIteratorBitSet; + private final AtomicReferenceArray systemModificationIterator; + private final MessageHandler messageHandler; + + SystemQueue(DirectBitSet systemModificationIteratorBitSet, + AtomicReferenceArray systemModificationIterator, + MessageHandler messageHandler) { + + this.systemModificationIteratorBitSet = systemModificationIteratorBitSet; + this.systemModificationIterator = systemModificationIterator; + this.messageHandler = messageHandler; + } + } + + public class ChronicleChannel extends Replicator implements Closeable { + + private final short chronicleChannel; + + private ChronicleChannel(short chronicleChannel) { + this.chronicleChannel = chronicleChannel; + } + + public byte identifier() { + return localIdentifier; + } + + @Override + protected Closeable applyTo(AbstractChronicleMapBuilder builder, + Replica map, EntryExternalizable entryExternalizable, + final ChronicleMap chronicleMap) { + add(chronicleChannel, map, entryExternalizable); + return this; + } + + public ChannelProvider provider() { + return ChannelProvider.this; + } + + @Override + public void close() throws IOException { + chronicleChannelList.remove(chronicleChannels[chronicleChannel]); + chronicleChannelIds.remove(chronicleChannel); + chronicleChannels[chronicleChannel] = null; + channelEntryExternalizables[chronicleChannel] = null; + } + + } +} \ No newline at end of file diff --git a/src/main/java/net/openhft/chronicle/hash/ChannelProviderBuilder.java b/src/main/java/net/openhft/chronicle/hash/ChannelProviderBuilder.java new file mode 100644 index 000000000..d9a26b346 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/ChannelProviderBuilder.java @@ -0,0 +1,90 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash; + +import net.openhft.chronicle.map.TcpReplicator; +import net.openhft.chronicle.map.UdpReplicator; + +import java.io.IOException; + +/** + * @author Rob Austin. + */ +public final class ChannelProviderBuilder { + + byte identifier; + int maxEntrySize = 1024; + int maxNumberOfChronicles = 128; + private UdpReplicationConfig udpReplicationConfig = null; + private TcpReplicationConfig tcpReplicationConfig = null; + + + + public int maxEntrySize() { + return maxEntrySize; + } + + public ChannelProviderBuilder maxEntrySize(int maxEntrySize) { + this.maxEntrySize = maxEntrySize; + return this; + } + + public ChannelProviderBuilder replicators(byte identifier, ReplicationConfig... replicationConfigs) { + this.identifier = identifier; + + + for (ReplicationConfig replicationConfig : replicationConfigs) { + + if (replicationConfig instanceof TcpReplicationConfig) { + this.tcpReplicationConfig = (TcpReplicationConfig) replicationConfig; + } else if (replicationConfig instanceof UdpReplicationConfig) { + this.udpReplicationConfig = (UdpReplicationConfig) replicationConfig; + } else + throw new UnsupportedOperationException(); + } + return this; + } + + public ChannelProviderBuilder maxNumberOfChronicles(int maxNumberOfChronicles) { + this.maxNumberOfChronicles = maxNumberOfChronicles; + return this; + } + + public ChannelProvider create() throws IOException { + final ChannelProvider replicatingCluster = new ChannelProvider(this); + if (tcpReplicationConfig != null) { + final TcpReplicator tcpReplicator = new TcpReplicator( + replicatingCluster.asReplica, + replicatingCluster.asEntryExternalizable, + tcpReplicationConfig, + maxEntrySize, null); + replicatingCluster.add(tcpReplicator); + } + + if (udpReplicationConfig != null) { + final UdpReplicator udpReplicator = + new UdpReplicator(replicatingCluster.asReplica, + replicatingCluster.asEntryExternalizable, + udpReplicationConfig, + maxEntrySize); + replicatingCluster.add(udpReplicator); + } + return replicatingCluster; + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/ReplicationConfig.java b/src/main/java/net/openhft/chronicle/hash/ReplicationConfig.java new file mode 100644 index 000000000..edb7e9269 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/ReplicationConfig.java @@ -0,0 +1,27 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash; + +/** + * @author Rob Austin. + */ +public interface ReplicationConfig { + + +} diff --git a/src/main/java/net/openhft/chronicle/hash/StatelessBuilder.java b/src/main/java/net/openhft/chronicle/hash/StatelessBuilder.java new file mode 100644 index 000000000..5434eb768 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/StatelessBuilder.java @@ -0,0 +1,52 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash; + +import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; + +/** + * @author Rob Austin. + */ +public class StatelessBuilder { + + private long timeoutMs = TimeUnit.SECONDS.toMillis(10); + + private InetSocketAddress remoteAddress; + + public long timeoutMs() { + return timeoutMs; + } + + public void timeout(long timeout, TimeUnit units) { + this.timeoutMs = units.toMillis(timeout); + } + + private StatelessBuilder(InetSocketAddress remoteAddress) { + this.remoteAddress = remoteAddress; + } + + public static StatelessBuilder remoteAddress(InetSocketAddress inetSocketAddress) { + return new StatelessBuilder(inetSocketAddress); + } + + public InetSocketAddress remoteAddress() { + return remoteAddress; + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/TcpReplicationConfig.java b/src/main/java/net/openhft/chronicle/hash/TcpReplicationConfig.java new file mode 100644 index 000000000..22e4aa6de --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/TcpReplicationConfig.java @@ -0,0 +1,207 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash; + +import net.openhft.chronicle.map.IdentifierListener; + +import org.jetbrains.annotations.NotNull; + +import java.net.InetSocketAddress; +import java.util.Arrays; +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.TimeUnit; + +import static java.util.Collections.unmodifiableSet; +import static java.util.concurrent.TimeUnit.SECONDS; + + +public class TcpReplicationConfig implements ReplicationConfig { + + private static final int DEFAULT_PACKET_SIZE = 1024 * 8; + private static final long DEFAULT_HEART_BEAT_INTERVAL = 20; + private static final TimeUnit DEFAULT_HEART_BEAT_INTERVAL_UNIT = SECONDS; + + private int serverPort = DEFAULT_PACKET_SIZE; + private Set endpoints; + private int packetSize = DEFAULT_PACKET_SIZE; + private boolean autoReconnectedUponDroppedConnection = true; + private ThrottlingConfig throttlingConfig = ThrottlingConfig.noThrottling(); + private long heartBeatInterval = DEFAULT_HEART_BEAT_INTERVAL; + private TimeUnit heartBeatIntervalUnit = DEFAULT_HEART_BEAT_INTERVAL_UNIT; + public IdentifierListener identifierListener; + + TcpReplicationConfig() { + + } + + public IdentifierListener identifierListener() { + return identifierListener; + } + + public TcpReplicationConfig nonUniqueIdentifierListener(@NotNull final IdentifierListener + identifierListener) { + this.identifierListener = identifierListener; + return this; + } + + public boolean autoReconnectedUponDroppedConnection() { + return autoReconnectedUponDroppedConnection; + } + + public TcpReplicationConfig autoReconnectedUponDroppedConnection(boolean autoReconnectedUponDroppedConnection) { + this.autoReconnectedUponDroppedConnection = autoReconnectedUponDroppedConnection; + return this; + } + + public ThrottlingConfig throttlingConfig() { + return throttlingConfig; + } + + public TcpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) { + this.throttlingConfig = throttlingConfig; + return this; + } + + public long heartBeatInterval(TimeUnit unit) { + return unit.convert(heartBeatInterval, heartBeatIntervalUnit); + } + + public int serverPort() { + return serverPort; + } + + public TcpReplicationConfig serverPort(int serverPort) { + this.serverPort = serverPort; + return this; + } + + public Set endpoints() { + return endpoints; + } + + public TcpReplicationConfig endpoints(Set endpoints) { + + for (final InetSocketAddress endpoint : endpoints) { + if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName())) + throw new IllegalArgumentException("endpoint=" + endpoint + + " can not point to the same port as the server"); + } + + this.endpoints = endpoints; + return this; + } + + public int packetSize() { + return packetSize; + } + + public static TcpReplicationConfig of(int serverPort, InetSocketAddress... endpoints) { + return of(serverPort, Arrays.asList(endpoints)); + } + + public static TcpReplicationConfig of(int serverPort, Collection endpoints) { + // at least in tests, we configure "receive-only" replication without endpoints. + // TODO decide what to do with this case +// if (endpoints.isEmpty()) +// throw new IllegalArgumentException("There should be some endpoints"); + for (final InetSocketAddress endpoint : endpoints) { + if (endpoint.getPort() == serverPort && "localhost".equals(endpoint.getHostName())) + throw new IllegalArgumentException("endpoint=" + endpoint + + " can not point to the same port as the server"); + } + + TcpReplicationConfig tcpReplicationConfig = new TcpReplicationConfig(); + + tcpReplicationConfig.serverPort(serverPort); + tcpReplicationConfig.endpoints = unmodifiableSet(new HashSet(endpoints)); + + return tcpReplicationConfig; + + + } + + + public TcpReplicationConfig packetSize(int packetSize) { + if (packetSize <= 0) + throw new IllegalArgumentException(); + this.packetSize = packetSize; + return this; + } + + public TcpReplicationConfig throttlingConfig(long heartBeatInterval, TimeUnit heartBeatIntervalUnit) { + this.heartBeatInterval = heartBeatInterval; + this.heartBeatIntervalUnit = heartBeatIntervalUnit; + return this; + } + + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + TcpReplicationConfig that = (TcpReplicationConfig) o; + + if (autoReconnectedUponDroppedConnection != that.autoReconnectedUponDroppedConnection) return false; + if (heartBeatInterval != that.heartBeatInterval) return false; + if (packetSize != that.packetSize) return false; + if (serverPort != that.serverPort) return false; + if (endpoints != null ? !endpoints.equals(that.endpoints) : that.endpoints != null) return false; + if (heartBeatIntervalUnit != that.heartBeatIntervalUnit) return false; + if (throttlingConfig != null ? !throttlingConfig.equals(that.throttlingConfig) : that.throttlingConfig != null) + return false; + + return true; + } + + @Override + public int hashCode() { + int result = serverPort; + result = 31 * result + (endpoints != null ? endpoints.hashCode() : 0); + result = 31 * result + packetSize; + result = 31 * result + (autoReconnectedUponDroppedConnection ? 1 : 0); + result = 31 * result + (throttlingConfig != null ? throttlingConfig.hashCode() : 0); + result = 31 * result + (int) (heartBeatInterval ^ (heartBeatInterval >>> 32)); + result = 31 * result + (heartBeatIntervalUnit != null ? heartBeatIntervalUnit.hashCode() : 0); + return result; + } + + @Override + public String toString() { + return "TcpReplicationConfig{" + + "serverPort=" + serverPort + + ", endpoints=" + endpoints + + ", packetSize=" + packetSize + + ", autoReconnectedUponDroppedConnection=" + autoReconnectedUponDroppedConnection + + ", throttlingConfig=" + throttlingConfig + + ", heartBeatInterval=" + heartBeatInterval + + ", heartBeatIntervalUnit=" + heartBeatIntervalUnit + + '}'; + } + + public TcpReplicationConfig heartBeatInterval(long time, TimeUnit unit) { + this.heartBeatInterval = time; + this.heartBeatIntervalUnit = unit; + return this; + } + + +} diff --git a/src/main/java/net/openhft/chronicle/hash/ThrottlingConfig.java b/src/main/java/net/openhft/chronicle/hash/ThrottlingConfig.java new file mode 100644 index 000000000..dbb466e2d --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/ThrottlingConfig.java @@ -0,0 +1,173 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash; + + +import java.util.concurrent.TimeUnit; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; + + +public class ThrottlingConfig { + + private static final long DEFAULT_BUCKET_INTERVAL = 100L; + private static final TimeUnit DEFAULT_BUCKET_INTERVAL_UNIT = MILLISECONDS; + + private static final ThrottlingConfig NO_THROTTLING = create(0L, MILLISECONDS, + DEFAULT_BUCKET_INTERVAL, DEFAULT_BUCKET_INTERVAL_UNIT); + + private final long throttle; + private final TimeUnit throttlePerUnit; + private final long bucketInterval; + private final TimeUnit bucketIntervalUnit; + + + ThrottlingConfig( + 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; + } + + /** + * Returns a config which prescribe that a replicator shouldn't throttle. + * + * @return a config which prescribe that a replicator shouldn't throttle + */ + public static ThrottlingConfig noThrottling() { + return NO_THROTTLING; + } + + /** + * 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 + * @return a config which throttle the specified max bits per the given time unit + * @throws IllegalArgumentException if the specified {@code maxBits} is non-positive value + */ + public static ThrottlingConfig throttle(long maxBits, TimeUnit perUnit) { + if (maxBits <= 0L) + throw new IllegalArgumentException(); + return create(maxBits, perUnit, DEFAULT_BUCKET_INTERVAL, DEFAULT_BUCKET_INTERVAL_UNIT); + } + + public static void checkMillisecondBucketInterval(ThrottlingConfig config, String replicator) { + if (config.bucketInterval(MILLISECONDS) < 1) { + throw new IllegalArgumentException( + "Minimum throttle bucketing interval of 1 millisecond is supported by " + + replicator + " replicator"); + } + } + + static ThrottlingConfig create( + long throttle, TimeUnit throttlePerUnit, + long throttleBucketInterval, TimeUnit throttleBucketIntervalUnit) { + return new ThrottlingConfig(throttle, throttlePerUnit, + throttleBucketInterval, throttleBucketIntervalUnit); + } + + + /** + * 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 + */ + public final long throttling(TimeUnit perUnit) { + return throttlePerUnit.convert(throttle, perUnit); + } + + + /** + * Returns the throttle bucketing interval in the given time units.

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 + */ + public long bucketInterval(TimeUnit unit) { + return unit.convert(bucketInterval, bucketIntervalUnit); + } + + /** + * Returns a copy of this config with the specified bucketing interval. + * + * @param throttleBucketInterval the bucketing interval for throttling + * @param unit the time unit of the interval + * @return a copy of this config with the specified bucketing interval + * @throws IllegalArgumentException if the given bucketing interval is non-positive + */ + public ThrottlingConfig bucketInterval(long throttleBucketInterval, TimeUnit unit) { + if (throttleBucketInterval <= 0L) + throw new IllegalArgumentException(); + return create(throttle, throttlePerUnit, throttleBucketInterval, unit); + } + + @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; + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/TimeProvider.java b/src/main/java/net/openhft/chronicle/hash/TimeProvider.java new file mode 100644 index 000000000..bdf1f565c --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/TimeProvider.java @@ -0,0 +1,74 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash; + +import net.openhft.chronicle.map.ChronicleMap; +import net.openhft.chronicle.set.ChronicleSet; + +import java.io.ObjectStreamException; +import java.io.Serializable; + +/** + * TimeProvider was aims to possibly later provide an optimization to + * {@link System#currentTimeMillis()} on every call to replicated {@link ChronicleMap} + * or {@link ChronicleSet}. + * + * Subclasses should be immutable, because {@link ChronicleHashBuilder} doesn't make defensive + * copies. + * + * @author Rob Austin. + * @see ChronicleHashBuilder#timeProvider(TimeProvider) + */ +public abstract class TimeProvider implements Serializable { + /** + * Delegates {@link #currentTimeMillis()} to {@link System#currentTimeMillis()}. + */ + public static final TimeProvider SYSTEM = new System(); + private static final long serialVersionUID = 0L; + + public abstract long currentTimeMillis(); + + private static class System extends TimeProvider { + private static final long serialVersionUID = 0L; + + public long currentTimeMillis() { + return java.lang.System.currentTimeMillis(); + } + + @Override + public boolean equals(Object o) { + return o != null && o.getClass() == getClass(); + } + + @Override + public int hashCode() { + return getClass().hashCode(); + } + + @Override + public String toString() { + return getClass().getSimpleName(); + } + + private Object readResolve() throws ObjectStreamException { + return SYSTEM; + } + } + +} diff --git a/src/main/java/net/openhft/chronicle/hash/UdpReplicationConfig.java b/src/main/java/net/openhft/chronicle/hash/UdpReplicationConfig.java new file mode 100644 index 000000000..85076edcb --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/UdpReplicationConfig.java @@ -0,0 +1,157 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +import java.net.InetAddress; +import java.net.NetworkInterface; + + +public abstract class UdpReplicationConfig implements ReplicationConfig { + + /** + * Package-private constructor forbids subclassing from outside of the package + */ + UdpReplicationConfig() { + // nothing to do + } + + public static UdpReplicationConfig simple(@NotNull InetAddress address, int port) { + if (address.isMulticastAddress()) + throw new IllegalArgumentException(); + + + return create(address, port, null, ThrottlingConfig.noThrottling()); + } + + public static UdpReplicationConfig multiCast(@NotNull InetAddress address, int port, + @NotNull NetworkInterface networkInterface) { + if (!address.isMulticastAddress() || networkInterface == null) + throw new IllegalArgumentException(); + return create(address, port, networkInterface, ThrottlingConfig.noThrottling()); + } + + static UdpReplicationConfig create(InetAddress address, int port, + NetworkInterface networkInterface, + ThrottlingConfig throttlingConfig) { + return new UdpReplicationConfigBean(address, port, networkInterface, + throttlingConfig); + } + + public abstract InetAddress address(); + + public abstract int port(); + + @Nullable + public abstract NetworkInterface networkInterface(); + + public abstract ThrottlingConfig throttlingConfig(); + + public UdpReplicationConfig throttlingConfig(ThrottlingConfig throttlingConfig) { + ThrottlingConfig.checkMillisecondBucketInterval(throttlingConfig, "UDP"); + return create(address(), port(), networkInterface(), throttlingConfig); + } +} + + + +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; + } +} diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/AgileBytesMarshaller.java b/src/main/java/net/openhft/chronicle/hash/serialization/AgileBytesMarshaller.java new file mode 100644 index 000000000..6ab1f7b15 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/serialization/AgileBytesMarshaller.java @@ -0,0 +1,22 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash.serialization; + +public interface AgileBytesMarshaller extends BytesReader, BytesInterop, SizeMarshaller { +} diff --git a/src/main/java/net/openhft/chronicle/hash/serialization/BytesReaders.java b/src/main/java/net/openhft/chronicle/hash/serialization/BytesReaders.java new file mode 100644 index 000000000..6d6f89202 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/hash/serialization/BytesReaders.java @@ -0,0 +1,51 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.hash.serialization; + +import net.openhft.lang.io.Bytes; +import net.openhft.lang.io.serialization.BytesMarshaller; + +public final class BytesReaders { + + public static BytesReader fromBytesMarshaller(BytesMarshaller marshaller) { + return new SimpleBytesReader(marshaller); + } + + private static class SimpleBytesReader implements BytesReader { + private static final long serialVersionUID = 0L; + + private final BytesMarshaller marshaller; + + public SimpleBytesReader(BytesMarshaller marshaller) { + this.marshaller = marshaller; + } + + @Override + public E read(Bytes bytes, long size) { + return marshaller.read(bytes); + } + + @Override + public E read(Bytes bytes, long size, E e) { + return marshaller.read(bytes, e); + } + } + + private BytesReaders() {} +} diff --git a/src/main/java/net/openhft/chronicle/map/IdentifierListener.java b/src/main/java/net/openhft/chronicle/map/IdentifierListener.java new file mode 100644 index 000000000..d1b12f369 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/IdentifierListener.java @@ -0,0 +1,38 @@ +/* + * Copyright 2014 Higher Frequency Trading + * + * http://www.higherfrequencytrading.com + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package net.openhft.chronicle.map; + +import java.net.SocketAddress; + +/** + * @author Rob Austin. + */ +public interface IdentifierListener { + + /** + * checks the identifier that is unique and we haven't seen it before, unless it comes from the same port + * and host. + * + * @param remoteIdentifier remoteIdentifier + * @param remoteAddress remoteAddress + * @return true if unique + */ + boolean isIdentifierUnique(byte remoteIdentifier, SocketAddress remoteAddress); + +} diff --git a/target/test-classes/war_and_peace.txt.gz b/target/test-classes/war_and_peace.txt.gz new file mode 100644 index 000000000..16131fa4b Binary files /dev/null and b/target/test-classes/war_and_peace.txt.gz differ