From 0f0c928676dd32dd67f5d8551853ca3134a73b8f Mon Sep 17 00:00:00 2001 From: Rob Austin Date: Wed, 26 Nov 2014 18:09:31 +0000 Subject: [PATCH] HCOLL-216 introduced Netty reflection patch for NIO selectors --- .../map/AbstractChannelReplicator.java | 28 +++++++------ .../chronicle/map/StatelessChronicleMap.java | 4 +- .../openhft/chronicle/map/TcpReplicator.java | 40 ++++++++++++------- 3 files changed, 44 insertions(+), 28 deletions(-) diff --git a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java index 3558ba343..52d4a57ab 100644 --- a/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/AbstractChannelReplicator.java @@ -51,27 +51,29 @@ abstract class AbstractChannelReplicator implements Closeable { public static final int BITS_IN_A_BYTE = 8; public static final int SIZE_OF_SIZE = 4; - public static final int SIZE_OF_TRANSACTIONID = 8; + public static final int SIZE_OF_TRANSACTION_ID = 8; private static final Logger LOG = LoggerFactory.getLogger(AbstractChannelReplicator.class); - final Selector selector; + + // currently this is not safe to use as it wont work with the stateless client + static boolean useJavaNIOSelectionKeys = Boolean.valueOf(System.getProperty("useJavaNIOSelectionKeys")); + + final SelectedSelectionKeySet selectedKeys = new SelectedSelectionKeySet(); + final CloseablesManager closeables = new CloseablesManager(); + final Selector selector = openSelector(closeables); private final ExecutorService executorService; private final Queue pendingRegistrations = new ConcurrentLinkedQueue(); @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) throws IOException { executorService = Executors.newSingleThreadExecutor( new NamedThreadFactory(name, true)); - selector = openSelector(closeables); throttler = throttlingConfig.throttling(DAYS) > 0 ? new Throttler(selector, @@ -84,7 +86,7 @@ abstract class AbstractChannelReplicator implements Closeable { Selector result = Selector.open(); closeables.add(result); - if (useOptimizedSelectedKeys) { + if (!useJavaNIOSelectionKeys) { closeables.add(new Closeable() { @Override public void close() throws IOException { @@ -105,7 +107,7 @@ abstract class AbstractChannelReplicator implements Closeable { } ); - return openSelector(result); + return openSelector(result, selectedKeys); } @@ -137,11 +139,13 @@ abstract class AbstractChannelReplicator implements Closeable { * @param selector * @return */ - private Selector openSelector(Selector selector) { + private Selector openSelector(@NotNull final Selector selector, @NotNull final + SelectedSelectionKeySet + selectedKeySet) { try { - SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet(); + Class selectorImplClass = Class.forName("sun.nio.ch.SelectorImpl", false, getSystemClassLoader()); @@ -154,17 +158,17 @@ abstract class AbstractChannelReplicator implements Closeable { Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys"); Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys"); + selectedKeysField.setAccessible(true); publicSelectedKeysField.setAccessible(true); + Object old = selectedKeysField.get(selector); 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); } diff --git a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java index c487150ba..0c0e6c297 100644 --- a/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java +++ b/src/main/java/net/openhft/chronicle/map/StatelessChronicleMap.java @@ -42,7 +42,7 @@ import java.util.concurrent.*; import static java.nio.ByteBuffer.allocateDirect; import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTIONID; +import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTION_ID; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.*; /** @@ -888,7 +888,7 @@ class StatelessChronicleMap implements ChronicleMap, Closeable, Clon long sizeLocation = writeEvent(event); // skips for the transaction id - bytes.skip(SIZE_OF_TRANSACTIONID); + bytes.skip(SIZE_OF_TRANSACTION_ID); return sizeLocation; } diff --git a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java index 45f3470a0..3fbc6525e 100644 --- a/src/main/java/net/openhft/chronicle/map/TcpReplicator.java +++ b/src/main/java/net/openhft/chronicle/map/TcpReplicator.java @@ -42,7 +42,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import static java.nio.channels.SelectionKey.*; import static java.util.concurrent.TimeUnit.MILLISECONDS; import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_SIZE; -import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTIONID; +import static net.openhft.chronicle.map.AbstractChannelReplicator.SIZE_OF_TRANSACTION_ID; import static net.openhft.chronicle.map.StatelessChronicleMap.EventId.HEARTBEAT; interface Work { @@ -142,7 +142,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable registerPendingRegistrations(); - final int nSelectedKeys = select(); + final int nSelectedKeys = selector.selectNow(); // its less resource intensive to set this less frequently and use an approximation final long approxTime = System.currentTimeMillis(); @@ -155,8 +155,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable // set the OP_WRITE when data is ready to send opWriteUpdater.applyUpdates(); - - if (selectedKeys == null) { + if (useJavaNIOSelectionKeys) { // use the standard java nio selector @@ -173,10 +172,9 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable final SelectionKey[] keys = selectedKeys.flip(); - if (keys.length == 0) - continue; + try { - for (int i = 0; i< keys.length && keys[i] != null; i++) { + for (int i = 0; i < keys.length && keys[i] != null; i++) { final SelectionKey key = keys[i]; try { @@ -188,11 +186,12 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } } finally { - for (int i = 0; i< keys.length && keys[i] != null; i++) { + for (int i = 0; i < keys.length && keys[i] != null; i++) { keys[i] = null; } } + // Thread.sleep(100); } } @@ -229,7 +228,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable if (key.isAcceptable()) { if (LOG.isDebugEnabled()) - LOG.info("onAccept - " + name); + LOG.debug("onAccept - " + name); onAccept(key); } @@ -240,9 +239,13 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable } if (key.isReadable()) { + if (LOG.isDebugEnabled()) + LOG.debug("onRead - " + name); onRead(key, approxTime); } if (key.isWritable()) { + if (LOG.isDebugEnabled()) + LOG.debug("onWrite - " + name); onWrite(key, approxTime); } } catch (BufferUnderflowException e) { @@ -495,7 +498,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable channel.socket().setSoLinger(false, 0); final Attached attached = new Attached(); - channel.register(selector, OP_WRITE | OP_READ, attached); + channel.register(selector, OP_READ, attached); throttle(channel); @@ -534,8 +537,11 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable final TcpSocketChannelEntryWriter writer = attached.entryWriter; final TcpSocketChannelEntryReader reader = attached.entryReader; + socketChannel.register(selector, OP_READ | OP_WRITE, attached); + if (attached.remoteIdentifier == Byte.MIN_VALUE) { + final byte remoteIdentifier = reader.identifierFromBuffer(); if (remoteIdentifier == STATELESS_CLIENT) { @@ -641,9 +647,6 @@ 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, @@ -652,6 +655,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable if (bytesJustWritten > 0) contemplateThrottleWrites(bytesJustWritten); + if (!attached.entryWriter.hasBytesToWrite() && !attached.entryWriter.isWorkIncomplete()) + // TURN OP_WRITE_OFF + key.interestOps(key.interestOps() & ~OP_WRITE); + } catch (IOException e) { quietClose(key, e); if (!attached.isServer) @@ -663,6 +670,7 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable /** * called when the selector receives a OP_READ message */ + private void onRead(final SelectionKey key, final long approxTime) throws IOException, InterruptedException { @@ -1122,6 +1130,10 @@ final class TcpReplicator extends AbstractChannelReplicator implements Closeable public boolean doWork() { return uncompletedWork.doWork(in); } + + public boolean hasBytesToWrite() { + return in.position() > 0; + } } /** @@ -1338,7 +1350,7 @@ class StatelessServerConnector { public static final StatelessChronicleMap.EventId[] VALUES = StatelessChronicleMap.EventId.values(); public static final int SIZE_OF_IS_EXCEPTION = 1; - public static final int HEADER_SIZE = SIZE_OF_SIZE + SIZE_OF_IS_EXCEPTION + SIZE_OF_TRANSACTIONID; + public static final int HEADER_SIZE = SIZE_OF_SIZE + SIZE_OF_IS_EXCEPTION + SIZE_OF_TRANSACTION_ID; private final KeyValueSerializer keyValueSerializer;