From b747788152ffac0e9bceb0da9a5fd7f226bc4385 Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Tue, 25 Nov 2014 20:45:43 +0000 Subject: [PATCH] HCOLL-216 added a fix to the TcpReplicator to stop OP_WRITE from spinning when working with the stateless client - also introduced the Netty like code for patching the nio selector to reduce object creation. --- .../map/AbstractChannelReplicator.java | 71 +++++++- .../map/SelectedSelectionKeySet.java | 110 ++++++++++++ .../openhft/chronicle/map/TcpReplicator.java | 157 +++++++++++++----- .../chronicle/map/SingleMapChannelTest.java | 7 +- .../map/TCPSocketReplication4WayMapTest.java | 12 +- 5 files changed, 302 insertions(+), 55 deletions(-) create mode 100644 src/main/java/net/openhft/chronicle/map/SelectedSelectionKeySet.java diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index a18173bdc..a503333d3 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -29,8 +29,11 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; import java.io.IOException; +import java.lang.reflect.Field; import java.net.InetSocketAddress; import java.nio.channels.*; +import java.security.AccessController; +import java.security.PrivilegedAction; import java.util.Queue; import java.util.Set; import java.util.concurrent.*; @@ -58,6 +61,10 @@ abstract class AbstractChannelReplicator implements Closeable { @Nullable private final Throttler throttler; volatile boolean isClosed = false; + SelectedSelectionKeySet selectedKeys; + + // currently this is not safe to use as it wont work with the stateless client + private boolean useOptimizedSelectedKeys = Boolean.valueOf(System.getProperty("useOptimizedSelectedKeys")); AbstractChannelReplicator(String name, ThrottlingConfig throttlingConfig, int maxEntrySizeBytes) @@ -72,10 +79,16 @@ abstract class AbstractChannelReplicator implements Closeable { maxEntrySizeBytes, throttlingConfig.throttling(DAYS)) : null; } - static Selector openSelector(final CloseablesManager closeables) throws IOException { + Selector openSelector(final CloseablesManager closeables) throws IOException { Selector result = null; try { - result = Selector.open(); + if (useOptimizedSelectedKeys) { + Selector selector = Selector.open(); + closeables.add(selector); + result = openSelector(selector); + } else + result = Selector.open(); + } finally { if (result != null) closeables.add(result); @@ -101,6 +114,60 @@ abstract class AbstractChannelReplicator implements Closeable { return result; } + + /** + * this is similar to the code in Netty + * + * @param selector + * @return + */ + private Selector openSelector(Selector selector) { + + + try { + SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet(); + + Class selectorImplClass = + Class.forName("sun.nio.ch.SelectorImpl", false, getSystemClassLoader()); + + // Ensure the current selector implementation is what we can instrument. + if (!selectorImplClass.isAssignableFrom(selector.getClass())) { + return selector; + } + + Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys"); + Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys"); + + selectedKeysField.setAccessible(true); + publicSelectedKeysField.setAccessible(true); + + selectedKeysField.set(selector, selectedKeySet); + publicSelectedKeysField.set(selector, selectedKeySet); + + this.selectedKeys = selectedKeySet; + // logger.trace("Instrumented an optimized java.util.Set into: {}", selector); + } catch (Exception e) { + LOG.error("", e); + this.selectedKeys = null; + // logger.trace("Failed to instrument an optimized java.util.Set into: {}", selector, t); + } + + return selector; + } + + static ClassLoader getSystemClassLoader() { + if (System.getSecurityManager() == null) { + return ClassLoader.getSystemClassLoader(); + } else { + return AccessController.doPrivileged(new PrivilegedAction() { + @Override + public ClassLoader run() { + return ClassLoader.getSystemClassLoader(); + } + }); + } + } + void addPendingRegistration(Runnable registration) { pendingRegistrations.add(registration); } diff --git a/src/main/java/net/openhft/chronicle/map/SelectedSelectionKeySet.java b/src/main/java/net/openhft/chronicle/map/SelectedSelectionKeySet.java new file mode 100644 index 000000000..4a29b31de --- /dev/null +++ b/src/main/java/net/openhft/chronicle/map/SelectedSelectionKeySet.java @@ -0,0 +1,110 @@ +/* + * Copyright 2013 The Netty Project + * + * The Netty Project licenses this file to you 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 java.nio.channels.SelectionKey; +import java.util.AbstractSet; +import java.util.Iterator; + +final class SelectedSelectionKeySet extends AbstractSet { + + private SelectionKey[] keysA; + private int keysASize; + private SelectionKey[] keysB; + private int keysBSize; + private boolean isA = true; + + SelectedSelectionKeySet() { + keysA = new SelectionKey[1024]; + keysB = keysA.clone(); + } + + @Override + public boolean add(SelectionKey o) { + if (o == null) { + return false; + } + + if (isA) { + int size = keysASize; + keysA[size ++] = o; + keysASize = size; + if (size == keysA.length) { + doubleCapacityA(); + } + } else { + int size = keysBSize; + keysB[size ++] = o; + keysBSize = size; + if (size == keysB.length) { + doubleCapacityB(); + } + } + + return true; + } + + private void doubleCapacityA() { + SelectionKey[] newKeysA = new SelectionKey[keysA.length << 1]; + System.arraycopy(keysA, 0, newKeysA, 0, keysASize); + keysA = newKeysA; + } + + private void doubleCapacityB() { + SelectionKey[] newKeysB = new SelectionKey[keysB.length << 1]; + System.arraycopy(keysB, 0, newKeysB, 0, keysBSize); + keysB = newKeysB; + } + + SelectionKey[] flip() { + if (isA) { + isA = false; + keysA[keysASize] = null; + keysBSize = 0; + return keysA; + } else { + isA = true; + keysB[keysBSize] = null; + keysASize = 0; + return keysB; + } + } + + @Override + public int size() { + if (isA) { + return keysASize; + } else { + return keysBSize; + } + } + + @Override + public boolean remove(Object o) { + return false; + } + + @Override + public boolean contains(Object o) { + return false; + } + + @Override + public Iterator iterator() { + throw new UnsupportedOperationException(); + } +} diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index ee6aa4675..f850b058c 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -32,6 +32,7 @@ import org.slf4j.LoggerFactory; import java.io.Closeable; import java.io.IOException; import java.net.*; +import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.nio.channels.*; @@ -65,6 +66,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable public static final byte NOT_SET = (byte) HEARTBEAT.ordinal(); private static final Logger LOG = LoggerFactory.getLogger(TcpReplicator.class.getName()); private static final int BUFFER_SIZE = 0x100000; // 1MB + public static final int SPIN_LOOP_COUNT = 100000; 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 @@ -140,7 +142,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable registerPendingRegistrations(); - final int nSelectedKeys = selector.select(selectorTimeout); + final int nSelectedKeys = select(); // its less resource intensive to set this less frequently and use an approximation final long approxTime = System.currentTimeMillis(); @@ -153,53 +155,42 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // set the OP_WRITE when data is ready to send opWriteUpdater.applyUpdates(); - if (nSelectedKeys == 0) - continue; // go back and check pendingRegistrations - final Set selectionKeys = selector.selectedKeys(); - for (final SelectionKey key : selectionKeys) { - try { + if (selectedKeys == null) { - if (!key.isValid()) - continue; + // use the standard java nio selector - if (key.isAcceptable()) { - if (LOG.isDebugEnabled()) - LOG.info("onAccept - " + name); - onAccept(key); - } + if (nSelectedKeys == 0) + continue; // go back and check pendingRegistrations - if (key.isConnectable()) { - if (LOG.isDebugEnabled()) - LOG.debug("onConnect - " + name); - onConnect(key); - } - - if (key.isReadable()) - onRead(key, approxTime); - - if (key.isWritable()) - onWrite(key, approxTime); - - } catch (CancelledKeyException e) { - if (!isClosed) - quietClose(key, e); - } catch (ClosedSelectorException e) { - if (!isClosed) - quietClose(key, e); - } catch (IOException e) { - if (!isClosed) - quietClose(key, e); - } catch (InterruptedException e) { - if (!isClosed) - quietClose(key, e); - } catch (Exception e) { - LOG.info("", e); - if (!isClosed) - closeEarlyAndQuietly(key.channel()); + final Set selectionKeys = selector.selectedKeys(); + for (final SelectionKey key : selectionKeys) { + processKey(approxTime, key); } + selectionKeys.clear(); + } else { + // use the netty like selector + + final SelectionKey[] keys = selectedKeys.flip(); + + if (keys.length == 0) + continue; + + 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); + } + + } + } - selectionKeys.clear(); + } } catch (CancelledKeyException e) { if (LOG.isDebugEnabled()) @@ -216,6 +207,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) { @@ -224,6 +216,73 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } } + private void processKey(long approxTime, SelectionKey key) { + + try { + + if (!key.isValid()) + return; + + if (key.isAcceptable()) { + if (LOG.isDebugEnabled()) + LOG.info("onAccept - " + name); + onAccept(key); + } + + if (key.isConnectable()) { + if (LOG.isDebugEnabled()) + LOG.debug("onConnect - " + name); + onConnect(key); + } + + if (key.isReadable()) { + onRead(key, approxTime); + } + if (key.isWritable()) { + onWrite(key, approxTime); + } + } catch (BufferUnderflowException e) { + if (!isClosed) + quietClose(key, e); + } catch (CancelledKeyException e) { + if (!isClosed) + quietClose(key, e); + } catch (ClosedSelectorException e) { + if (!isClosed) + quietClose(key, e); + } catch (IOException e) { + if (!isClosed) + quietClose(key, e); + } catch (InterruptedException e) { + if (!isClosed) + quietClose(key, e); + } catch (Exception e) { + LOG.info("", e); + if (!isClosed) + closeEarlyAndQuietly(key.channel()); + } + + } + + /** + * spin loops 100000 times first before calling the selector with timeout + * + * @return + * @throws IOException + */ + private int select() throws IOException { + + // spin loop 100000 times + for (int i = 0; i < SPIN_LOOP_COUNT; i++) { + final int keys = selector.selectNow(); + if (keys != 0) + return keys; + } + + return selector.select(selectorTimeout); + } + + /** * checks that we receive heartbeats and send out heart beats. * @@ -557,7 +616,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // now we're finished we can get on with reading the entries attached.handShakingComplete = true; attached.remoteModificationIterator.dirtyEntries(attached.remoteBootstrapTimestamp); - reader.entriesFromBuffer(attached); + reader.entriesFromBuffer(attached, key); } } @@ -578,6 +637,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } else if (attached.remoteModificationIterator != null) attached.entryWriter.entriesToBuffer(attached.remoteModificationIterator, key); + if (attached.isHandShakingComplete() && !attached.entryWriter.isWorkIncomplete()) + // TURN OP_WRITE_OFF + key.interestOps(key.interestOps() & ~OP_WRITE); try { final int bytesJustWritten = attached.entryWriter.writeBufferToSocket(socketChannel, @@ -635,7 +697,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable attached.entryReader.lastHeartBeatReceived = approxTime; if (attached.isHandShakingComplete()) { - attached.entryReader.entriesFromBuffer(attached); + attached.entryReader.entriesFromBuffer(attached, key); } else { doHandShaking(key, socketChannel); } @@ -974,6 +1036,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * * @param socketChannel the socket to publish the buffer to * @param approxTime an approximation of the current time in millis + * @param key the selectors key * @throws IOException */ private int writeBufferToSocket(@NotNull final SocketChannel socketChannel, @@ -1127,9 +1190,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable * reads entries from the buffer till empty * * @param attached + * @param key * @throws InterruptedException */ - void entriesFromBuffer(Attached attached) throws InterruptedException, IOException { + void entriesFromBuffer(Attached attached, SelectionKey key) throws InterruptedException, IOException { for (; ; ) { out.limit(in.position()); @@ -1180,6 +1244,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable final Work futureWork = statelessServerConnector.processStatelessEvent(state, attached.entryWriter.in, attached.entryReader.out); + // turn the OP_WRITE on + key.interestOps(key.interestOps() | OP_WRITE); + // in some cases it may not be possible to send out all the data before we // fill out the write buffer, so this data will be send when the buffer // is no longer full, and as such is treated as future work diff --git a/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java b/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java index 315cd86e6..ac5062742 100644 --- a/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java +++ b/src/test/java/net/openhft/chronicle/map/SingleMapChannelTest.java @@ -62,9 +62,10 @@ public class SingleMapChannelTest { { final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(8086, new InetSocketAddress("localhost", 8087)) + .autoReconnectedUponDroppedConnection(true) .heartBeatInterval(1, SECONDS); int maxEntrySize = 1024; - byte identifier= 1; + byte identifier = 1; hubA = ReplicationHub.builder() .maxEntrySize(maxEntrySize) .tcpTransportAndNetwork(tcpConfig) @@ -79,9 +80,11 @@ public class SingleMapChannelTest { { final TcpTransportAndNetworkConfig tcpConfig = TcpTransportAndNetworkConfig .of(8087, new InetSocketAddress("localhost", 8086)) + .autoReconnectedUponDroppedConnection(true) .heartBeatInterval(1, SECONDS); - hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig).createWithId((byte) 2); + hubB = ReplicationHub.builder().tcpTransportAndNetwork(tcpConfig).createWithId( + (byte) 2); // this is how you add maps after the custer is created map1b = ChronicleMapBuilder.of(Integer.class, CharSequence.class) diff --git a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java index 51e9f3831..38ce5cde0 100644 --- a/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java +++ b/src/test/java/net/openhft/chronicle/map/TCPSocketReplication4WayMapTest.java @@ -19,7 +19,6 @@ package net.openhft.chronicle.map; -import net.openhft.lang.values.IntValue; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -147,11 +146,12 @@ public class TCPSocketReplication4WayMapTest { private void waitTillEqual(final int timeOutMs) throws InterruptedException { int t = 0; for (; t < timeOutMs; t++) { - if (map1.equals(map2) && - map1.equals(map3) && - map2.equals(map3) && - map4.equals(map3) && - map1.equals(map4)) + if ( map1.equals(map2) && + map1.equals(map3) && + map1.equals(map4) && + map2.equals(map3) && + map2.equals(map4) && + map3.equals(map4)) break; Thread.sleep(1); }