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; }