From 69ea009f02dd5f8fdf7244d6a8e0106e1d412b28 Mon Sep 17 00:00:00 2001 From: leventov Date: Sun, 14 Sep 2014 15:00:34 +0400 Subject: [PATCH] HCOLL-133 Avoid duplicate closing of TCP/UDP replicators --- .../map/AbstractChannelReplicator.java | 2 + .../chronicle/map/CloseablesManager.java | 2 +- .../openhft/chronicle/map/TcpReplicator.java | 15 +-- .../chronicle/map/UdpChannelReplicator.java | 96 ++++++++++--------- 4 files changed, 57 insertions(+), 58 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 36edc6cd6..deade5e8d 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -58,6 +58,7 @@ abstract class AbstractChannelReplicator implements Closeable { private final Queue pendingRegistrations = new ConcurrentLinkedQueue(); @Nullable private final Throttler throttler; + volatile boolean isClosed = false; AbstractChannelReplicator(String name, ThrottlingConfig throttlingConfig, int maxEntrySizeBytes) @@ -108,6 +109,7 @@ abstract class AbstractChannelReplicator implements Closeable { @Override public void close() { + isClosed = true; closeables.closeQuietly(); executorService.shutdownNow(); } diff --git a/src/main/java/net/openhft/chronicle/map/CloseablesManager.java b/src/main/java/net/openhft/chronicle/map/CloseablesManager.java index 1355c5920..e4a655530 100644 --- a/src/main/java/net/openhft/chronicle/map/CloseablesManager.java +++ b/src/main/java/net/openhft/chronicle/map/CloseablesManager.java @@ -26,7 +26,7 @@ import java.util.Iterator; import java.util.List; final class CloseablesManager implements Closeable { - private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName()); + private static final Logger LOG = LoggerFactory.getLogger(CloseablesManager.class.getName()); private boolean isClosed = false; private List closeables = new ArrayList(); diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index f9da96941..bd73f8202 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -176,17 +176,10 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable { LOG.debug("", e); } catch (Exception e) { LOG.error("", e); - } finally - - { - if (selector != null) - try { - selector.close(); - } catch (IOException e) { - if (LOG.isDebugEnabled()) - LOG.debug("", e); - } - close(); + } finally { + if (!isClosed) { + close(); + } } } diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java index 5e48c6fc9..6330bc714 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -92,8 +92,8 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. @Override public void close() { - writeChannel = null; super.close(); + writeChannel = null; } /** @@ -105,59 +105,63 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. connectClient().register(selector, OP_READ); serverConnector.connectLater(); - while (selector.isOpen()) { + try { + while (selector.isOpen()) { - registerPendingRegistrations(); + 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); + // 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(); + if (shouldEnableOpWrite) + enableWrites(); - checkThrottleInterval(); + 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()); + 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(); + } + } finally { + if (!isClosed) { + close(); } - - selectionKeys.clear(); } - - }