From 41a39fe22852896c6f0edc8fa8861e891714ab07 Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 10 Sep 2014 00:02:14 +0400 Subject: [PATCH 1/3] HCOLL-128 simplified ReplicatingCluster: removed unnecessary use of DirectBitSet --- .../chronicle/map/ReplicatingCluster.java | 50 +++++++++++-------- 1 file changed, 30 insertions(+), 20 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java b/src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java index f3d7344b3..87d31d036 100644 --- a/src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java +++ b/src/main/java/net/openhft/chronicle/map/ReplicatingCluster.java @@ -28,6 +28,8 @@ 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; @@ -93,10 +95,12 @@ public final class ReplicatingCluster implements Closeable { @Override public boolean hasNext() { - for (int i = (int) chronicleChannelBitSet.nextSetBit(0); i >= 0; i = (int) - chronicleChannelBitSet.nextSetBit(i + 1)) { - final ModificationIterator modificationIterator = chronicleChannels[i] - .acquireModificationIterator(remoteIdentifier, notifier); + // 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; } @@ -106,11 +110,11 @@ public final class ReplicatingCluster implements Closeable { @Override public boolean nextEntry(@NotNull EntryCallback callback, final int na) { - for (int i = (int) chronicleChannelBitSet.nextSetBit(0); i >= 0; - i = (int) chronicleChannelBitSet.nextSetBit(i + 1)) { - final ModificationIterator modificationIterator = chronicleChannels[i] + 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, i)) + if (modificationIterator.nextEntry(callback, chronicleChannelIds.get(i))) return true; } return false; @@ -118,9 +122,9 @@ public final class ReplicatingCluster implements Closeable { @Override public void dirtyEntries(long fromTimeStamp) { - for (int i = (int) chronicleChannelBitSet.nextSetBit(0); i >= 0; - i = (int) chronicleChannelBitSet.nextSetBit(i + 1)) { - chronicleChannels[i].acquireModificationIterator(remoteIdentifier, notifier) + for (int i = 0, len = chronicleChannelList.size(); i < len; i++) { + chronicleChannelList.get(i) + .acquireModificationIterator(remoteIdentifier, notifier) .dirtyEntries(fromTimeStamp); notifier.onChange(); } @@ -138,8 +142,8 @@ public final class ReplicatingCluster implements Closeable { @Override public long lastModificationTime(byte remoteIdentifier) { long t = 0; - for (int i = (int) chronicleChannelBitSet.previousSetBit(chronicleChannels.length); - i > 0; i = (int) chronicleChannelBitSet.previousSetBit(i - 1)) { + // 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)); } @@ -152,8 +156,9 @@ public final class ReplicatingCluster implements Closeable { } }; private final int maxEntrySize; - private final DirectBitSet chronicleChannelBitSet; private final Replica[] chronicleChannels; + private final List chronicleChannelList; + private final List chronicleChannelIds; private final EntryExternalizable[] channelEntryExternalizables; private final AtomicReferenceArray systemModificationIterator = new AtomicReferenceArray(128); @@ -169,7 +174,8 @@ public final class ReplicatingCluster implements Closeable { maxEntrySize = builder.maxEntrySize; chronicleChannels = new Replica[builder.maxNumberOfChronicles]; channelEntryExternalizables = new EntryExternalizable[builder.maxNumberOfChronicles]; - chronicleChannelBitSet = newBitSet(chronicleChannels.length); + chronicleChannelList = new ArrayList(); + chronicleChannelIds = new ArrayList(); MessageHandler systemMessageHandler = new MessageHandler() { @Override public void onMessage(Bytes bytes) { @@ -241,8 +247,9 @@ public final class ReplicatingCluster implements Closeable { " is already in use."); } chronicleChannels[chronicleChannel] = replica; + chronicleChannelList.add(replica); + chronicleChannelIds.add((int) chronicleChannel); channelEntryExternalizables[chronicleChannel] = entryExternalizable; - chronicleChannelBitSet.set(chronicleChannel); if (chronicleChannel == 0) return; @@ -263,9 +270,11 @@ public final class ReplicatingCluster implements Closeable { for (AbstractChannelReplicator replicator : replicators) { replicator.close(); } - for (int i = (int) chronicleChannelBitSet.nextSetBit(0); i > 0; - i = (int) chronicleChannelBitSet.nextSetBit(i + 1)) { - chronicleChannels[i].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(); } } @@ -408,7 +417,8 @@ public final class ReplicatingCluster implements Closeable { @Override public void close() throws IOException { - chronicleChannelBitSet.clear(chronicleChannel); + chronicleChannelList.remove(chronicleChannels[chronicleChannel]); + chronicleChannelIds.remove(chronicleChannel); chronicleChannels[chronicleChannel] = null; channelEntryExternalizables[chronicleChannel] = null; } From 2ef875ef91c0bbbbd67cc304408024185d6057f4 Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 10 Sep 2014 02:44:00 +0400 Subject: [PATCH 2/3] HCOLL-128 cleanup Replicators closeables handling, fixed small defect -- AbstractConnector.connectLater didn't remove channel from closeables --- .../map/AbstractChannelReplicator.java | 53 +++-------- .../chronicle/map/CloseablesManager.java | 87 +++++++++++++++++++ .../openhft/chronicle/map/TcpReplicator.java | 14 +-- .../openhft/chronicle/map/UdpReplicator.java | 27 +++--- 4 files changed, 113 insertions(+), 68 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/CloseablesManager.java diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 6d3c6e732..ce4d94227 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -55,7 +55,7 @@ abstract class AbstractChannelReplicator implements Closeable { static final int SIZE_OF_SHORT = 2; private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class); final Selector selector; - final Set closeables = Collections.synchronizedSet(new LinkedHashSet()); + final CloseablesManager closeables = new CloseablesManager(); private final ExecutorService executorService; private final Queue pendingRegistrations = new ConcurrentLinkedQueue(); @Nullable @@ -110,28 +110,14 @@ abstract class AbstractChannelReplicator implements Closeable { @Override public void close() { - synchronized (this.closeables) { - for (Closeable closeable : this.closeables) { - try { - closeable.close(); - } catch (IOException e) { - LOG.error("", e); - } - } - closeables.clear(); - } + closeables.closeQuietly(); executorService.shutdownNow(); } void closeEarlyAndQuietly(SelectableChannel channel) { - try { - if (throttler != null) - throttler.remove(channel); - closeables.remove(channel); - channel.close(); - } catch (IOException ex) { - // do nothing - } + if (throttler != null) + throttler.remove(channel); + closeables.closeQuietly(channel); } /** @@ -322,20 +308,15 @@ abstract class AbstractChannelReplicator implements Closeable { * connectionAttempts} */ public final void connectLater() { - try { - if (socketChannel != null) - socketChannel.close(); + if (socketChannel != null) { + closeables.closeQuietly(socketChannel); socketChannel = null; - - } catch (IOException e1) { - LOG.error("", e1); } final long reconnectionInterval = connectionAttempts * 100; if (connectionAttempts < 5) connectionAttempts++; doConnect(reconnectionInterval); - } /** @@ -354,31 +335,17 @@ abstract class AbstractChannelReplicator implements Closeable { final Thread thread = new Thread(new Runnable() { public void run() { - SelectableChannel socketChannel = null; try { if (reconnectionInterval > 0) Thread.sleep(reconnectionInterval); - synchronized (closeables) { - socketChannel = doConnect(); - if (socketChannel == null) - return; - - closeables.add(socketChannel); - AbstractConnector.this.socketChannel = socketChannel; - } + SelectableChannel socketChannel = doConnect(); + closeables.add(socketChannel); + AbstractConnector.this.socketChannel = socketChannel; } catch (Exception e) { - if (socketChannel != null) - try { - socketChannel.close(); - AbstractConnector.this.socketChannel = null; - } catch (IOException e1) { - // - } LOG.error("", e); } - } }); diff --git a/src/main/java/net/openhft/chronicle/map/CloseablesManager.java b/src/main/java/net/openhft/chronicle/map/CloseablesManager.java new file mode 100644 index 000000000..1355c5920 --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/CloseablesManager.java @@ -0,0 +1,87 @@ +/* + * 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.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.Closeable; +import java.io.IOException; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; + +final class CloseablesManager implements Closeable { + private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName()); + + private boolean isClosed = false; + private List closeables = new ArrayList(); + + private void checkState() { + if (isClosed) + throw new IllegalStateException(); + } + + public synchronized void add(Closeable closeable) { + checkState(); + if (closeable == null) + throw new NullPointerException(); + closeables.add(closeable); + } + + public synchronized void close(Closeable closeable) throws IOException { + checkState(); + for (Iterator iterator = closeables.iterator(); iterator.hasNext(); ) { + if (iterator.next() == closeable) + iterator.remove(); + } + closeable.close(); + } + + public synchronized void closeQuietly(Closeable closeable) { + try { + close(closeable); + } catch (IOException e) { + LOG.error("", e); + } + } + + @Override + public synchronized void close() throws IOException { + IOException ex = null; + for (Closeable closeable : closeables) { + try { + closeable.close(); + } catch (IOException e) { + LOG.error("", e); + ex = e; + } + } + closeables.clear(); + isClosed = true; + if (ex != null) + throw ex; + } + + public synchronized void closeQuietly() { + try { + close(); + } catch (IOException e) { + // do nothing + } + } +} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 1089cc697..b72051184 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -277,19 +277,13 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { final SocketChannel channel = (SocketChannel) key.channel(); - if (approxTimeOutTime > attached.entryReader.lastHeartBeatReceived + attached.remoteHeartbeatInterval) { + if (approxTimeOutTime > + attached.entryReader.lastHeartBeatReceived + attached.remoteHeartbeatInterval) { if (LOG.isDebugEnabled()) LOG.debug("lost connection, attempting to reconnect. " + "missed heartbeat from identifier=" + attached.remoteIdentifier); - try { - channel.socket().close(); - channel.close(); - activeKeys.clear(attached.remoteIdentifier); - closeables.remove(channel); - } catch (IOException e) { - LOG.debug("", e); - } - + activeKeys.clear(attached.remoteIdentifier); + closeables.closeQuietly(channel.socket()); attached.connector.connectLater(); } } diff --git a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java index 193463b0b..3d91081ee 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpReplicator.java @@ -169,23 +169,20 @@ class UdpReplicator extends AbstractChannelReplicator implements Replica.Modific final InetSocketAddress hostAddress = new InetSocketAddress(port); client.configureBlocking(false); - synchronized (closeables) { - - 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 (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("Listening on port " + port); - closeables.add(client); + 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; } From d4ec1f2e27a4576c12d61bbac578b2407956e4c8 Mon Sep 17 00:00:00 2001 From: leventov Date: Wed, 10 Sep 2014 02:46:31 +0400 Subject: [PATCH 3/3] HCOLL-128 small TcpReplicator cleanup - removed pointless connectorBySocket mapping --- .../java/net/openhft/chronicle/map/TcpReplicator.java | 9 ++------- 1 file changed, 2 insertions(+), 7 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index b72051184..0dda8b663 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -46,7 +46,6 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName()); private static final int BUFFER_SIZE = 0x100000; // 1MB final AtomicBoolean forceBootstrap = new AtomicBoolean(); - private final Map connectorBySocket = new ConcurrentHashMap(); private final SelectionKey[] selectionKeysStore = new SelectionKey[Byte.MAX_VALUE + 1]; // used to instruct the selector thread to set OP_WRITE on a key correlated by the bit index in the // bitset @@ -101,15 +100,11 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { void process() throws IOException { try { final Details serverDetails = new Details(serverInetSocketAddress, localIdentifier); - connectorBySocket.put(serverInetSocketAddress, new ServerConnector(serverDetails)); + new ServerConnector(serverDetails).connect(); for (InetSocketAddress client : endpoints) { final Details clientDetails = new Details(client, localIdentifier); - connectorBySocket.put(client, new ClientConnector(clientDetails)); - } - - for (AbstractConnector connector : connectorBySocket.values()) { - connector.connect(); + new ClientConnector(clientDetails).connect(); } while (selector.isOpen()) {