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.

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<ClassLoader>() {
+                @Override
+                public ClassLoader run() {
+                    return ClassLoader.getSystemClassLoader();
+                }
+            });
+        }
+    }
+
     void addPendingRegistration(Runnable registration) {
         pendingRegistrations.add(registration);
     }

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<SelectionKey> {
+
+    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<SelectionKey> iterator() {
+        throw new UnsupportedOperationException();
+    }
+}

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<SelectionKey> 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<SelectionKey> 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