Alawn/Chronicle-Map
1
0
Fork 0
mirror of https://github.com/OpenHFT/Chronicle-Map.git synced 2021-09-13 01:26:12 +00:00
Code Releases Activity

HCOLL-133 Avoid duplicate closing of TCP/UDP replicators

Browse Source
This commit is contained in:
leventov
2014-09-14 15:00:34 +04:00
parent b0229b9afa
commit 69ea009f02
4 changed files with 57 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java
+2
View File
@@ -58,6 +58,7 @@ abstract class AbstractChannelReplicator implements Closeable {
private final Queue<Runnable> pendingRegistrations = new ConcurrentLinkedQueue<Runnable>();
@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();
}
src/main/java/net/openhft/chronicle/map/CloseablesManager.java
+1 -1
View File
@@ -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<Closeable> closeables = new ArrayList<Closeable>();
src/main/java/net/openhft/chronicle/map/TcpReplicator.java
+4 -11
View File
@@ -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();
}
}
}
src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java
+50 -46
View File
@@ -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<SelectionKey> 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<SelectionKey> 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();
}
}