From 8f80aaa1b8f58829da3f9b03c168002ef3f84887 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 26 Nov 2014 13:38:43 +0000 Subject: [PATCH] HCOLL-218 ReplicatedChronicleMapTest appears to use too much resources. reduced resources used by this test - ( plus a few other changes ) --- .../map/AbstractChannelReplicator.java | 42 +++++++++++++------ .../openhft/chronicle/map/TcpReplicator.java | 28 +++++++------ .../chronicle/map/UdpChannelReplicator.java | 2 +- .../chronicle/map/ChannelReplicationTest.java | 16 +++---- 4 files changed, 55 insertions(+), 33 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index a503333d3..3558ba343 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -80,19 +80,35 @@ abstract class AbstractChannelReplicator implements Closeable { } Selector openSelector(final CloseablesManager closeables) throws IOException { - Selector result = null; - try { - if (useOptimizedSelectedKeys) { - Selector selector = Selector.open(); - closeables.add(selector); - result = openSelector(selector); - } else - result = Selector.open(); - } finally { - if (result != null) - closeables.add(result); + Selector result = Selector.open(); + closeables.add(result); + + if (useOptimizedSelectedKeys) { + closeables.add(new Closeable() { + @Override + public void close() throws IOException { + { + SelectionKey[] keys = selectedKeys.flip(); + for (int i = 0; i < keys.length && keys[i] != null; i++) { + keys[i] = null; + } + } + { + SelectionKey[] keys = selectedKeys.flip(); + for (int i = 0; i < keys.length && keys[i] != null; i++) { + keys[i] = null; + } + } + } + + + } + ); + return openSelector(result); + } + return result; } @@ -183,7 +199,7 @@ abstract class AbstractChannelReplicator implements Closeable { } } - abstract void process() throws IOException; + abstract void processEvent() throws IOException; final void start() { executorService.execute( @@ -191,7 +207,7 @@ abstract class AbstractChannelReplicator implements Closeable { @Override public void run() { try { - process(); + processEvent(); } catch (Exception e) { LOG.error("", e); } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index f850b058c..45f3470a0 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -126,7 +126,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } @Override - void process() throws IOException { + void processEvent() throws IOException { try { final InetSocketAddress serverInetSocketAddress = new InetSocketAddress(replicationConfig .serverPort()); @@ -175,18 +175,22 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable if (keys.length == 0) continue; + try { + for (int i = 0; i< keys.length && keys[i] != null; i++) { + final SelectionKey key = keys[i]; + + try { + processKey(approxTime, key); + } catch (BufferUnderflowException e) { + if (!isClosed) + LOG.error("", e); + } - for (int i = 0; ; i++) { - final SelectionKey key = keys[i]; - if (key == null) - break; - try { - processKey(approxTime, key); - } catch (BufferUnderflowException e) { - if (!isClosed) - LOG.error("", e); } - + } finally { + for (int i = 0; i< keys.length && keys[i] != null; i++) { + keys[i] = null; + } } } @@ -207,7 +211,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } catch (Exception e) { LOG.error("", e); } finally { - selectedKeys = null; + if (LOG.isDebugEnabled()) LOG.debug("closing name=" + name); if (!isClosed) { diff --git a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java index 0a1b25946..cf7a7e5f7 100644 --- a/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/UdpChannelReplicator.java @@ -104,7 +104,7 @@ class UdpChannelReplicator extends AbstractChannelReplicator implements Replica. * binds to the server socket and process data This method will block until interrupted */ @Override - void process() throws IOException { + void processEvent() throws IOException { connectClient().register(selector, OP_READ); serverConnector.connectLater(); diff --git a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java index 5405a6795..dff303bf2 100644 --- a/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java +++ b/src/test/java/net/openhft/chronicle/map/ChannelReplicationTest.java @@ -46,7 +46,6 @@ public class ChannelReplicationTest { private ReplicationHub hubA; private ReplicationHub hubB; - @Before public void setup() throws IOException { { @@ -59,7 +58,8 @@ public class ChannelReplicationTest { .createWithId((byte) 1); map1a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(1000) + .entries(2) + .entrySize(100) .instance().replicatedViaChannel(hubA.createChannel((short) 1)).create(); } @@ -72,7 +72,8 @@ public class ChannelReplicationTest { .createWithId((byte) 2); map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(1000) + .entries(2) + .entrySize(100) .instance().replicatedViaChannel(hubB.createChannel((short) 1)).create(); } } @@ -90,17 +91,18 @@ public class ChannelReplicationTest { } - @Ignore + @Test public void test() throws IOException, InterruptedException { map2b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(1000) + .entries(2) + .entrySize(100) .instance().replicatedViaChannel(hubB.createChannel((short) 2)).create(); - map2a = ChronicleMapBuilder.of(Integer.class, CharSequence.class) - .entries(1000) + .entries(2) + .entrySize(100) .instance().replicatedViaChannel(hubA.createChannel((short) 2)).create(); map2a.put(1, "EXAMPLE-2");