From b0229b9afac73523bbc235ee61aa465c5ccd086b Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Sat, 13 Sep 2014 09:19:32 +0100 Subject: [PATCH] HCOLL-134 Broadcast active ChronicleMap host names over UDP to allow other maps to connect --- .../map/AbstractChannelReplicator.java | 3 - .../chronicle/map/ChronicleMapBuilder.java | 1 - .../chronicle/map/UdpChannelReplicator.java | 291 ++++++++++++++++++ .../openhft/chronicle/map/UdpReplicator.java | 259 ++-------------- 4 files changed, 317 insertions(+), 237 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index ecbbdad9e..36edc6cd6 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -31,8 +31,6 @@ import java.nio.channels.ClosedChannelException; import java.nio.channels.SelectableChannel; import java.nio.channels.SelectionKey; import java.nio.channels.Selector; -import java.util.Collections; -import java.util.LinkedHashSet; import java.util.Queue; import java.util.Set; import java.util.concurrent.ConcurrentLinkedQueue; @@ -186,7 +184,6 @@ abstract class AbstractChannelReplicator implements Closeable { lastTime = time; bytesWritten = 0; - if (LOG.isDebugEnabled()) LOG.debug("Restoring OP_WRITE on all channels"); diff --git a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java index 1decb8a7e..3925471cd 100644 --- a/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java +++ b/src/main/java/net/openhft/chronicle/map/ChronicleMapBuilder.java @@ -28,7 +28,6 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.*; -import java.nio.channels.FileChannel; import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeUnit; diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java new file mode 100644 index 000000000..5e48c6fc9 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -0,0 +1,291 @@ +/* + * 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 org.jetbrains.annotations.NotNull; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.NetworkInterface; +import java.net.StandardSocketOptions; +import java.nio.channels.*; +import java.util.Set; + +import static java.net.StandardProtocolFamily.INET; +import static java.net.StandardProtocolFamily.INET6; +import static java.net.StandardSocketOptions.IP_MULTICAST_IF; +import static java.net.StandardSocketOptions.SO_REUSEADDR; +import static java.nio.channels.SelectionKey.OP_READ; +import static java.nio.channels.SelectionKey.OP_WRITE; + + +/** + * The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ), typically, 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. In order to not miss data, UdpReplicator should be used in conjunction with + * the TCP Replicator. + */ +class UdpChannelReplicator extends AbstractChannelReplicator implements Replica.ModificationNotifier, Closeable { + + private static final Logger LOG = + LoggerFactory.getLogger(UdpChannelReplicator.class.getName()); + + private final byte localIdentifier; + private EntryWriter writer; + private EntryReader reader; + + private final InetAddress address; + private final int port; + private final NetworkInterface networkInterface; + private final ServerConnector serverConnector; + + private SelectableChannel writeChannel; + private volatile boolean shouldEnableOpWrite; + + /** + * @param replicationConfig + * @param maxPayloadSize the maximum size of any serialized entry + * @param localIdentifier just used for logging + * @throws IOException + */ + UdpChannelReplicator(@NotNull final UdpReplicationConfig replicationConfig, + final int maxPayloadSize, final byte localIdentifier) + throws IOException { + + super("UdpReplicator-" + localIdentifier, replicationConfig.throttlingConfig(), + maxPayloadSize); + + this.localIdentifier = localIdentifier; + + address = replicationConfig.address(); + port = replicationConfig.port(); + networkInterface = replicationConfig.networkInterface(); + serverConnector = new ServerConnector(); + } + + + void setWriter(EntryWriter writer) { + this.writer = writer; + } + + void setReader(EntryReader reader) { + this.reader = reader; + } + + @Override + public void close() { + writeChannel = null; + super.close(); + } + + /** + * binds to the server socket and process data This method will block until interrupted + */ + @Override + void process() throws IOException { + + connectClient().register(selector, OP_READ); + serverConnector.connectLater(); + + while (selector.isOpen()) { + + registerPendingRegistrations(); + + // this may block for a long time, upon return the + // selected set contains keys of the ready channels + final int n = selector.select(100); + + if (shouldEnableOpWrite) + enableWrites(); + + checkThrottleInterval(); + + if (n == 0) { + continue; // nothing to do + } + + final Set selectionKeys = selector.selectedKeys(); + for (final SelectionKey key : selectionKeys) { + + try { + + if (key.isReadable()) { + final DatagramChannel socketChannel = (DatagramChannel) key.channel(); + reader.readAll(socketChannel); + } + + if (key.isWritable()) { + final DatagramChannel socketChannel = (DatagramChannel) key.channel(); + try { + int bytesJustWritten = writer.writeAll(socketChannel); + contemplateThrottleWrites(bytesJustWritten); + } catch (NotYetConnectedException e) { + if (LOG.isDebugEnabled()) + LOG.debug("", e); + serverConnector.connectLater(); + } catch (IOException e) { + if (LOG.isDebugEnabled()) + LOG.debug("", e); + serverConnector.connectLater(); + } + } + + } catch (Exception e) { + LOG.error("", e); + closeEarlyAndQuietly(key.channel()); + } + } + + selectionKeys.clear(); + } + + + } + + + private DatagramChannel connectClient() throws IOException { + final DatagramChannel client = address.isMulticastAddress() ? + DatagramChannel.open(address.getAddress().length == 4 ? INET : INET6) : + DatagramChannel.open(); + + final InetSocketAddress hostAddress = new InetSocketAddress(port); + client.configureBlocking(false); + if (address.isMulticastAddress()) { + client.setOption(IP_MULTICAST_IF, networkInterface); + client.setOption(SO_REUSEADDR, true); + client.bind(hostAddress); + client.join(address, networkInterface); + if (LOG.isDebugEnabled()) + LOG.debug("Connecting via multicast, group=" + address); + } else { + client.bind(hostAddress); + } + + if (LOG.isDebugEnabled()) + LOG.debug("Listening on port " + port); + closeables.add(client); + return client; + } + + /** + * called whenever there is a change to the modification iterator + */ + @Override + public void onChange() { + // the write have to be enabled on the same thread as the selector + shouldEnableOpWrite = true; + selector.wakeup(); + } + + private void enableWrites() { + + if (writeChannel == null) + return; + + try { + final SelectionKey selectionKey = writeChannel.keyFor(this.selector); + if (selectionKey != null) + selectionKey.interestOps(selectionKey.interestOps() | OP_WRITE); + } catch (Exception e) { + LOG.error("", e); + } + } + + void disableWrites() { + + if (writeChannel == null) + return; + + try { + final SelectionKey selectionKey = writeChannel.keyFor(this.selector); + selectionKey.interestOps(selectionKey.interestOps() & ~OP_WRITE); + } catch (Exception e) { + LOG.error("", e); + } + } + + + private class ServerConnector extends AbstractConnector { + private final InetSocketAddress socketAddress; + + private ServerConnector() { + super("UDP-Connector"); + this.socketAddress = new InetSocketAddress(address, port); + } + + SelectableChannel doConnect() throws + IOException, InterruptedException { + + final DatagramChannel server = DatagramChannel.open(); + server.configureBlocking(false); + + // Kick off connection establishment + try { + // Create a non-blocking socket channel + server.socket().setBroadcast(true); + server.connect(socketAddress); + } catch (IOException e) { + if (LOG.isDebugEnabled()) + LOG.debug("details=" + new Details(socketAddress, localIdentifier), e); + connectLater(); + return null; + } + + server.setOption(SO_REUSEADDR, true) + .setOption(StandardSocketOptions.IP_MULTICAST_LOOP, false) + .setOption(StandardSocketOptions.SO_BROADCAST, true) + .setOption(SO_REUSEADDR, true); + + // the registration has be be run on the same thread as the selector + addPendingRegistration(new Runnable() { + @Override + public void run() { + + try { + server.register(selector, OP_WRITE); + writeChannel = server; + throttle(server); + } catch (ClosedChannelException e) { + LOG.error("", e); + } + + } + }); + + return server; + } + + + } + +} + + +interface EntryReader { + void readAll(@NotNull final DatagramChannel socketChannel) throws IOException, InterruptedException; +} + +interface EntryWriter { + int writeAll(@NotNull final DatagramChannel socketChannel) throws InterruptedException, IOException; +} + + + diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 6c4faef0a..927c8edc8 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -23,203 +23,43 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; import java.io.IOException; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.net.NetworkInterface; -import java.net.StandardSocketOptions; import java.nio.ByteBuffer; -import java.nio.channels.*; -import java.util.Set; - -import static java.net.StandardProtocolFamily.INET; -import static java.net.StandardProtocolFamily.INET6; -import static java.net.StandardSocketOptions.IP_MULTICAST_IF; -import static java.net.StandardSocketOptions.SO_REUSEADDR; -import static java.nio.channels.SelectionKey.OP_READ; -import static java.nio.channels.SelectionKey.OP_WRITE; +import java.nio.channels.DatagramChannel; /** - * The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ), - * typically, 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. - * In order to not miss data, UdpReplicator should be used in conjunction with the TCP Replicator. + * The UdpReplicator attempts to read the data ( but it does not enforce or grantee delivery ), typically, 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. In order to not miss data, UdpReplicator should be used in conjunction with + * the TCP Replicator. */ -class UdpReplicator extends AbstractChannelReplicator implements Replica.ModificationNotifier, Closeable { +class UdpReplicator extends UdpChannelReplicator implements Replica.ModificationNotifier, Closeable { private static final Logger LOG = LoggerFactory.getLogger(UdpReplicator.class.getName()); - private final byte localIdentifier; - private final UdpSocketChannelEntryWriter writer; - private final UdpSocketChannelEntryReader reader; - private final Replica replica; - private final InetAddress address; - private final int port; - private final NetworkInterface networkInterface; - private final ServerConnector serverConnector; - private Replica.ModificationIterator modificationIterator; - private SelectableChannel writeChannel; - private volatile boolean shouldEnableOpWrite; UdpReplicator(@NotNull final Replica replica, @NotNull final Replica.EntryExternalizable entryExternalizable, @NotNull final UdpReplicationConfig replicationConfig, final int serializedEntrySize) throws IOException { - super("UdpReplicator-" + replica.identifier(), replicationConfig.throttlingConfig(), - serializedEntrySize); - this.localIdentifier = replica.identifier(); - this.replica = replica; - this.writer = new UdpSocketChannelEntryWriter(serializedEntrySize, entryExternalizable); - this.reader = new UdpSocketChannelEntryReader(serializedEntrySize, entryExternalizable); + super(replicationConfig, serializedEntrySize, replica.identifier()); - address = replicationConfig.address(); - port = replicationConfig.port(); - networkInterface = replicationConfig.networkInterface(); - - serverConnector = new ServerConnector(); - - modificationIterator = replica.acquireModificationIterator( + Replica.ModificationIterator modificationIterator = replica.acquireModificationIterator( ChronicleMapBuilder.UDP_REPLICATION_MODIFICATION_ITERATOR_ID, this); + setReader(new UdpSocketChannelEntryReader(serializedEntrySize, entryExternalizable)); + + setWriter(new UdpSocketChannelEntryWriter(serializedEntrySize, entryExternalizable, + modificationIterator, this)); + start(); } - @Override - public void close() { - writeChannel = null; - super.close(); - } - /** - * binds to the server socket and process data This method will block until interrupted - */ - @Override - void process() throws IOException { - - connectClient().register(selector, OP_READ); - serverConnector.connectLater(); - - while (selector.isOpen()) { - - registerPendingRegistrations(); - - // this may block for a long time, upon return the - // selected set contains keys of the ready channels - final int n = selector.select(100); - - if (shouldEnableOpWrite) - enableWrites(); - - checkThrottleInterval(); - - if (n == 0) { - continue; // nothing to do - } - - final Set selectionKeys = selector.selectedKeys(); - for (final SelectionKey key : selectionKeys) { - - try { - - if (key.isReadable()) { - final DatagramChannel socketChannel = (DatagramChannel) key.channel(); - reader.readAll(socketChannel); - } - - if (key.isWritable()) { - final DatagramChannel socketChannel = (DatagramChannel) key.channel(); - try { - int bytesJustWritten = writer.writeAll(socketChannel); - contemplateThrottleWrites(bytesJustWritten); - } catch (NotYetConnectedException e) { - if (LOG.isDebugEnabled()) - LOG.debug("", e); - serverConnector.connectLater(); - } catch (IOException e) { - if (LOG.isDebugEnabled()) - LOG.debug("", e); - serverConnector.connectLater(); - } - } - - } catch (Exception e) { - LOG.error("", e); - closeEarlyAndQuietly(key.channel()); - } - } - - selectionKeys.clear(); - } - - - } - - - private DatagramChannel connectClient() throws IOException { - final DatagramChannel client = address.isMulticastAddress() ? - DatagramChannel.open(address.getAddress().length == 4 ? INET : INET6) : - DatagramChannel.open(); - - final InetSocketAddress hostAddress = new InetSocketAddress(port); - client.configureBlocking(false); - if (address.isMulticastAddress()) { - client.setOption(IP_MULTICAST_IF, networkInterface); - client.setOption(SO_REUSEADDR, true); - client.bind(hostAddress); - client.join(address, networkInterface); - if (LOG.isDebugEnabled()) - LOG.debug("Connecting via multicast, group=" + address); - } else { - client.bind(hostAddress); - } - - if (LOG.isDebugEnabled()) - LOG.debug("Listening on port " + port); - closeables.add(client); - return client; - } - - /** - * called whenever there is a change to the modification iterator - */ - @Override - public void onChange() { - // the write have to be enabled on the same thread as the selector - shouldEnableOpWrite = true; - selector.wakeup(); - } - - private void enableWrites() { - - if (writeChannel == null) - return; - - try { - final SelectionKey selectionKey = writeChannel.keyFor(this.selector); - if (selectionKey != null) - selectionKey.interestOps(selectionKey.interestOps() | OP_WRITE); - } catch (Exception e) { - LOG.error("", e); - } - } - - private void disableWrites() { - - if (writeChannel == null) - return; - - try { - final SelectionKey selectionKey = writeChannel.keyFor(this.selector); - selectionKey.interestOps(selectionKey.interestOps() & ~OP_WRITE); - } catch (Exception e) { - LOG.error("", e); - } - } - - private static class UdpSocketChannelEntryReader { + private static class UdpSocketChannelEntryReader implements EntryReader { private final Replica.EntryExternalizable externalizable; private final ByteBuffer in; @@ -246,7 +86,8 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific * @throws IOException * @throws InterruptedException */ - void readAll(@NotNull final DatagramChannel socketChannel) throws IOException, InterruptedException { + public void readAll(@NotNull final DatagramChannel socketChannel) throws IOException, + InterruptedException { out.clear(); in.clear(); @@ -276,20 +117,25 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific } - private class UdpSocketChannelEntryWriter { + private static class UdpSocketChannelEntryWriter implements EntryWriter { private final ByteBuffer out; private final ByteBufferBytes in; private final EntryCallback entryCallback; + private final UdpChannelReplicator udpReplicator; + private Replica.ModificationIterator modificationIterator; UdpSocketChannelEntryWriter(final int serializedEntrySize, - @NotNull final Replica.EntryExternalizable externalizable) { + @NotNull final Replica.EntryExternalizable externalizable, + @NotNull final Replica.ModificationIterator modificationIterator, + 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); entryCallback = new EntryCallback(externalizable, in); - + this.modificationIterator = modificationIterator; } /** @@ -308,7 +154,7 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific * @throws InterruptedException * @throws IOException */ - int writeAll(@NotNull final DatagramChannel socketChannel) + public int writeAll(@NotNull final DatagramChannel socketChannel) throws InterruptedException, IOException { out.clear(); @@ -318,7 +164,7 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific final boolean wasDataRead = modificationIterator.nextEntry(entryCallback, 0); if (!wasDataRead) { - disableWrites(); + udpReplicator.disableWrites(); return 0; } @@ -331,59 +177,6 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific } } - private class ServerConnector extends AbstractChannelReplicator.AbstractConnector { - private final InetSocketAddress socketAddress; - - private ServerConnector() { - super("UDP-Connector"); - this.socketAddress = new InetSocketAddress(address, port); - } - - SelectableChannel doConnect() throws - IOException, InterruptedException { - - final DatagramChannel server = DatagramChannel.open(); - server.configureBlocking(false); - - // Kick off connection establishment - try { - // Create a non-blocking socket channel - server.socket().setBroadcast(true); - server.connect(socketAddress); - } catch (IOException e) { - if (LOG.isDebugEnabled()) - LOG.debug("details=" + new Details(socketAddress, localIdentifier), e); - connectLater(); - return null; - } - - server.setOption(SO_REUSEADDR, true) - .setOption(StandardSocketOptions.IP_MULTICAST_LOOP, false) - .setOption(StandardSocketOptions.SO_BROADCAST, true) - .setOption(SO_REUSEADDR, true); - - // the registration has be be run on the same thread as the selector - addPendingRegistration(new Runnable() { - @Override - public void run() { - - try { - server.register(selector, OP_WRITE); - writeChannel = server; - throttle(server); - } catch (ClosedChannelException e) { - LOG.error("", e); - } - - } - }); - - return server; - } - - - } - }