HCOLL-167 when running the profile I found that we were not closing every connection, so no being super fussy about adding everything to closable.

2021-09-13 01:26:12 +00:00 · 2014-10-20 13:02:42 +01:00
parent 0043b16d5e
commit 023d12a6d0
6 changed files with 170 additions and 94 deletions
@@ -27,10 +27,7 @@ import org.slf4j.LoggerFactory;
 import java.io.Closeable;
 import java.io.IOException;
 import java.net.InetSocketAddress;
-import java.nio.channels.ClosedChannelException;
-import java.nio.channels.SelectableChannel;
-import java.nio.channels.SelectionKey;
-import java.nio.channels.Selector;
+import java.nio.channels.*;
 import java.util.Queue;
 import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -65,8 +62,7 @@ abstract class AbstractChannelReplicator implements Closeable {
            throws IOException {
        executorService = Executors.newSingleThreadExecutor(
                new NamedThreadFactory(name, true));
-        selector = Selector.open();
-        closeables.add(selector);
+        selector = openSelector();

        throttler = throttlingConfig.throttling(DAYS) > 0 ?
                new Throttler(selector,
@@ -74,6 +70,29 @@ abstract class AbstractChannelReplicator implements Closeable {
                        maxEntrySizeBytes, throttlingConfig.throttling(DAYS)) : null;
    }

+    Selector openSelector() throws IOException {
+        Selector result = null;
+        try {
+            result = Selector.open();
+        } finally {
+            if (result != null)
+                closeables.add(result);
+        }
+        return result;
+    }
+
+    SocketChannel openSocketChannel() throws IOException {
+        SocketChannel result = null;
+
+        try {
+            result = SocketChannel.open();
+        } finally {
+            if (result != null)
+                closeables.add(result);
+        }
+        return result;
+    }
+
    void addPendingRegistration(Runnable registration) {
        pendingRegistrations.add(registration);
    }
@@ -389,4 +408,5 @@ abstract class AbstractChannelReplicator implements Closeable {
        }
    }

+
 }
@@ -21,15 +21,15 @@ import org.slf4j.LoggerFactory;

 import java.io.Closeable;
 import java.io.IOException;
-import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.Iterator;
-import java.util.List;
+import java.util.Set;

 final class CloseablesManager implements Closeable {
    private static final Logger LOG = LoggerFactory.getLogger(CloseablesManager.class.getName());

    private boolean isClosed = false;
-    private final List<Closeable> closeables = new ArrayList<Closeable>();
+    private final Set<Closeable> closeables = new HashSet<Closeable>();

    private void checkState() {
        if (isClosed)
@@ -42,36 +42,39 @@ import static net.openhft.chronicle.serialization.Hasher.hash;
 import static net.openhft.lang.collection.DirectBitSet.NOT_FOUND;

 /**
- * A Replicating Multi Master HashMap <p/> <p>Each remote hash map, mirrors its changes over to another remote
- * hash map, neither hash map is considered the master store of data, each hash map uses timestamps to
- * reconcile changes. We refer to an instance of a remote hash-map as a node. A node will be connected to any
- * number of other nodes, for the first implementation the maximum number of nodes will be fixed. The data
- * that is stored locally in each node will become eventually consistent. So changes made to one node, for
- * example by calling put() will be replicated over to the other node. To achieve a high level of performance
- * and throughput, the call to put() won’t block, with concurrentHashMap, It is typical to check the return
- * code of some methods to obtain the old value for example remove(). Due to the loose coupling and lock free
- * nature of this multi master implementation,  this return value will only be the old value on the nodes
- * local data store. In other words the nodes are only concurrent locally. Its worth realising that another
- * node performing exactly the same operation may return a different value. However reconciliation will ensure
- * the maps themselves become eventually consistent. <p/> <p>Reconciliation <p/> <p>If two ( or more nodes )
- * were to receive a change to their maps for the same key but different values, say by a user of the maps,
- * calling the put(key, value). Then, initially each node will update its local store and each local store
- * will hold a different value, but the aim of multi master replication is to provide eventual consistency
- * across the nodes. So, with multi master when ever a node is changed it will notify the other nodes of its
- * change. We will refer to this notification as an event. The event will hold a timestamp indicating the time
- * the change occurred, it will also hold the state transition, in this case it was a put with a key and
- * value. Eventual consistency is achieved by looking at the timestamp from the remote node, if for a given
- * key, the remote nodes timestamp is newer than the local nodes timestamp, then the event from the remote
- * node will be applied to the local node, otherwise the event will be ignored. <p/> <p>However there is an
- * edge case that we have to concern ourselves with, If two nodes update their map at the same time with
- * different values, we have to deterministically resolve which update wins, because of eventual consistency
- * both nodes should end up locally holding the same data. Although it is rare two remote nodes could receive
- * an update to their maps at exactly the same time for the same key, we have to handle this edge case, its
- * therefore important not to rely on timestamps alone to reconcile the updates. Typically the update with the
- * newest timestamp should win, but in this example both timestamps are the same, and the decision made to one
- * node should be identical to the decision made to the other. We resolve this simple dilemma by using a node
- * identifier, each node will have a unique identifier, the update from the node with the smallest identifier
- * wins.
+ * A Replicating Multi Master HashMap <p/> <p>Each remote hash map, mirrors its changes over to
+ * another remote hash map, neither hash map is considered the master store of data, each hash map
+ * uses timestamps to reconcile changes. We refer to an instance of a remote hash-map as a node. A
+ * node will be connected to any number of other nodes, for the first implementation the maximum
+ * number of nodes will be fixed. The data that is stored locally in each node will become
+ * eventually consistent. So changes made to one node, for example by calling put() will be
+ * replicated over to the other node. To achieve a high level of performance and throughput, the
+ * call to put() won’t block, with concurrentHashMap, It is typical to check the return code of some
+ * methods to obtain the old value for example remove(). Due to the loose coupling and lock free
+ * nature of this multi master implementation,  this return value will only be the old value on the
+ * nodes local data store. In other words the nodes are only concurrent locally. Its worth realising
+ * that another node performing exactly the same operation may return a different value. However
+ * reconciliation will ensure the maps themselves become eventually consistent. <p/>
+ * <p>Reconciliation <p/> <p>If two ( or more nodes ) were to receive a change to their maps for the
+ * same key but different values, say by a user of the maps, calling the put(key, value). Then,
+ * initially each node will update its local store and each local store will hold a different value,
+ * but the aim of multi master replication is to provide eventual consistency across the nodes. So,
+ * with multi master when ever a node is changed it will notify the other nodes of its change. We
+ * will refer to this notification as an event. The event will hold a timestamp indicating the time
+ * the change occurred, it will also hold the state transition, in this case it was a put with a key
+ * and value. Eventual consistency is achieved by looking at the timestamp from the remote node, if
+ * for a given key, the remote nodes timestamp is newer than the local nodes timestamp, then the
+ * event from the remote node will be applied to the local node, otherwise the event will be
+ * ignored. <p/> <p>However there is an edge case that we have to concern ourselves with, If two
+ * nodes update their map at the same time with different values, we have to deterministically
+ * resolve which update wins, because of eventual consistency both nodes should end up locally
+ * holding the same data. Although it is rare two remote nodes could receive an update to their maps
+ * at exactly the same time for the same key, we have to handle this edge case, its therefore
+ * important not to rely on timestamps alone to reconcile the updates. Typically the update with the
+ * newest timestamp should win, but in this example both timestamps are the same, and the decision
+ * made to one node should be identical to the decision made to the other. We resolve this simple
+ * dilemma by using a node identifier, each node will have a unique identifier, the update from the
+ * node with the smallest identifier wins.
 *
 * @param <K> the entries key type
 * @param <V> the entries value type
@@ -103,7 +106,7 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,

        this.localIdentifier = builder.identifier();

-        if (localIdentifier==-1){
+        if (localIdentifier == -1) {
            throw new IllegalStateException("localIdentifier should not be -1");
        }

@@ -195,8 +198,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
    }

    /**
-     * Used in conjunction with map replication, all put events that originate from a remote node will be
-     * processed using this method.
+     * Used in conjunction with map replication, all put events that originate from a remote node
+     * will be processed using this method.
     *
     * @param key        key with which the specified value is to be associated
     * @param value      value to be associated with the specified key
@@ -268,8 +271,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
    }

    /**
-     * Used in conjunction with map replication, all remove events that originate from a remote node will be
-     * processed using this method.
+     * Used in conjunction with map replication, all remove events that originate from a remote node
+     * will be processed using this method.
     *
     * @param key        key with which the specified value is associated
     * @param value      value expected to be associated with the specified key
@@ -297,12 +300,6 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
            }
        }

-        try {
-            // give time for stuff to fully close
-            Thread.sleep(50);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
        super.close();

    }
@@ -362,8 +359,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
    }

    /**
-     * This method does not set a segment lock, A segment lock should be obtained before calling this method,
-     * especially when being used in a multi threaded context.
+     * This method does not set a segment lock, A segment lock should be obtained before calling
+     * this method, especially when being used in a multi threaded context.
     */
    @Override
    public void writeExternalEntry(@NotNull Bytes entry, @NotNull Bytes destination,
@@ -437,8 +434,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
    }

    /**
-     * This method does not set a segment lock, A segment lock should be obtained before calling this method,
-     * especially when being used in a multi threaded context.
+     * This method does not set a segment lock, A segment lock should be obtained before calling
+     * this method, especially when being used in a multi threaded context.
     */
    @Override
    public void readExternalEntry(@NotNull Bytes source) {
@@ -858,9 +855,10 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        }

        /**
-         * Used only with replication, its sometimes possible to receive an old ( or stale update ) from a
-         * remote map. This method is used to determine if we should ignore such updates. <p/> <p>We can
-         * reject put() and removes() when comparing times stamps with remote systems
+         * Used only with replication, its sometimes possible to receive an old ( or stale update )
+         * from a remote map. This method is used to determine if we should ignore such updates.
+         * <p/> <p>We can reject put() and removes() when comparing times stamps with remote
+         * systems
         *
         * @param entry      the maps entry
         * @param timestamp  the time the entry was created or updated
@@ -887,18 +885,19 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        }

        /**
-         * Puts entry. If {@code value} implements {@link Byteable} interface and {@code usingValue} is {@code
-         * true}, the value is backed with the bytes of this entry.
+         * Puts entry. If {@code value} implements {@link Byteable} interface and {@code usingValue}
+         * is {@code true}, the value is backed with the bytes of this entry.
         *
-         * @param hash2              a hash of the {@code keyBytes}. Caller was searching for the key in the
-         *                           {@code searchedHashLookup} using this hash.
+         * @param hash2              a hash of the {@code keyBytes}. Caller was searching for the
+         *                           key in the {@code searchedHashLookup} using this hash.
         * @param value              the value to put
-         * @param usingValue         {@code true} if the value should be backed with the bytes of the entry,
-         *                           if it implements {@link Byteable} interface, {@code false} if it should
-         *                           put itself
+         * @param usingValue         {@code true} if the value should be backed with the bytes of
+         *                           the entry, if it implements {@link Byteable} interface, {@code
+         *                           false} if it should put itself
         * @param identifier         the identifier of the outer CHM node
-         * @param timestamp          the timestamp when the entry was put <s>(this could be later if it was a
-         *                           remote put)</s> this method is called only from usual put or acquire
+         * @param timestamp          the timestamp when the entry was put <s>(this could be later if
+         *                           it was a remote put)</s> this method is called only from usual
+         *                           put or acquire
         * @param searchedHashLookup the hash lookup that used to find the entry based on the key
         * @return offset of the written entry in the Segment bytes
         * @see VanillaChronicleMap.Segment#putEntry
@@ -1200,8 +1199,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
    }

    /**
-     * receive an update from the map, via the MapEventListener and delegates the changes to the currently
-     * active modification iterators
+     * receive an update from the map, via the MapEventListener and delegates the changes to the
+     * currently active modification iterators
     */
    class ModificationDelegator extends MapEventListener<K, V, ChronicleMap<K, V>> {
        private static final long serialVersionUID = 0L;
@@ -1312,16 +1311,16 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        }

        /**
-         * Always throws {@code NotSerializableException} since instances of this class are not intended to be
-         * serializable.
+         * Always throws {@code NotSerializableException} since instances of this class are not
+         * intended to be serializable.
         */
        private void writeObject(ObjectOutputStream out) throws IOException {
            throw new NotSerializableException(getClass().getCanonicalName());
        }

        /**
-         * Always throws {@code NotSerializableException} since instances of this class are not intended to be
-         * serializable.
+         * Always throws {@code NotSerializableException} since instances of this class are not
+         * intended to be serializable.
         */
        private void readObject(ObjectInputStream in)
                throws IOException, ClassNotFoundException {
@@ -1332,14 +1331,15 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
    }

    /**
-     * Once a change occurs to a map, map replication requires that these changes are picked up by another
-     * thread, this class provides an iterator like interface to poll for such changes. <p/> <p>In most cases
-     * the thread that adds data to the node is unlikely to be the same thread that replicates the data over
-     * to the other nodes, so data will have to be marshaled between the main thread storing data to the map,
-     * and the thread running the replication. <p/> <p>One way to perform this marshalling, would be to pipe
-     * the data into a queue. However, This class takes another approach. It uses a bit set, and marks bits
-     * which correspond to the indexes of the entries that have changed. It then provides an iterator like
-     * interface to poll for such changes.
+     * Once a change occurs to a map, map replication requires that these changes are picked up by
+     * another thread, this class provides an iterator like interface to poll for such changes. <p/>
+     * <p>In most cases the thread that adds data to the node is unlikely to be the same thread that
+     * replicates the data over to the other nodes, so data will have to be marshaled between the
+     * main thread storing data to the map, and the thread running the replication. <p/> <p>One way
+     * to perform this marshalling, would be to pipe the data into a queue. However, This class
+     * takes another approach. It uses a bit set, and marks bits which correspond to the indexes of
+     * the entries that have changed. It then provides an iterator like interface to poll for such
+     * changes.
     *
     * @author Rob Austin.
     */
@@ -1376,7 +1376,8 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
         *
         * @param segmentIndex the index of the maps segment
         * @param pos          the position within this {@code segmentIndex}
-         * @return and index the has combined the {@code segmentIndex}  and  {@code pos} into a single value
+         * @return and index the has combined the {@code segmentIndex}  and  {@code pos} into a
+         * single value
         */
        private long combine(int segmentIndex, long pos) {
            return (((long) segmentIndex) << segmentIndexShift) | pos;
@@ -1418,8 +1419,9 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,


        /**
-         * you can continue to poll hasNext() until data becomes available. If are are in the middle of
-         * processing an entry via {@code nextEntry}, hasNext will return true until the bit is cleared
+         * you can continue to poll hasNext() until data becomes available. If are are in the middle
+         * of processing an entry via {@code nextEntry}, hasNext will return true until the bit is
+         * cleared
         *
         * @return true if there is an entry
         */
@@ -1487,16 +1489,16 @@ class ReplicatedChronicleMap<K, KI, MKI extends MetaBytesInterop<K, KI>,
        }

        /**
-         * Always throws {@code NotSerializableException} since instances of this class are not intended to be
-         * serializable.
+         * Always throws {@code NotSerializableException} since instances of this class are not
+         * intended to be serializable.
         */
        private void writeObject(ObjectOutputStream out) throws IOException {
            throw new NotSerializableException(getClass().getCanonicalName());
        }

        /**
-         * Always throws {@code NotSerializableException} since instances of this class are not intended to be
-         * serializable.
+         * Always throws {@code NotSerializableException} since instances of this class are not
+         * intended to be serializable.
         */
        private void readObject(ObjectInputStream in)
                throws IOException, ClassNotFoundException {
@@ -64,6 +64,8 @@ class StatelessMapClient<K, V> implements ChronicleMap<K, V> {
        this.keyValueSerializer = keyValueSerializer;

        clientChannel = SocketChannel.open();
+
+
        clientChannel.connect(remote);

        doHandShaking();
@@ -316,7 +316,14 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
    private void onConnect(@NotNull final SelectionKey key)
            throws IOException, InterruptedException {

-        final SocketChannel channel = (SocketChannel) key.channel();
+        SocketChannel channel = null;
+
+        try {
+            channel = (SocketChannel) key.channel();
+        } finally {
+            closeables.add(channel);
+        }
+
        final Attached attached = (Attached) key.attachment();

        try {
@@ -328,7 +335,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {

            // when node discovery is used ( by nodes broadcasting out their host:port over UDP ),
            // when new or restarted nodes are started up. they attempt to find the nodes
-            // on the grid by listening to the host and ports of the other nodes, so these nodes will establish the connection when they come back up,
+            // on the grid by listening to the host and ports of the other nodes,
+            // so these nodes will establish the connection when they come back up,
            // hence under these circumstances, polling a dropped node to attempt to reconnect is no-longer
            // required as the remote node will establish the connection its self on startup.
            if (replicationConfig.autoReconnectedUponDroppedConnection())
@@ -363,9 +371,25 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
     * called when the selector receives a OP_ACCEPT message
     */
    private void onAccept(@NotNull final SelectionKey key) throws IOException {
+        ServerSocketChannel server = null;
+
+        try {
+            server = (ServerSocketChannel) key.channel();
+        } finally {
+            if (server != null)
+                closeables.add(server);
+        }
+
+        SocketChannel channel = null;
+
+        try {
+            channel = server.accept();
+        } finally {
+            if (channel != null)
+                closeables.add(channel);
+        }
+

-        final ServerSocketChannel server = (ServerSocketChannel) key.channel();
-        final SocketChannel channel = server.accept();
        channel.configureBlocking(false);
        channel.socket().setReuseAddress(true);
        channel.socket().setTcpNoDelay(true);
@@ -631,11 +655,21 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
        SelectableChannel doConnect() throws
                IOException, InterruptedException {

-            final ServerSocketChannel serverChannel = ServerSocketChannel.open();
+
+            final ServerSocketChannel serverChannel = openServerSocketChannel();

            serverChannel.socket().setReceiveBufferSize(BUFFER_SIZE);
            serverChannel.configureBlocking(false);
-            final ServerSocket serverSocket = serverChannel.socket();
+            ServerSocket serverSocket = null;
+
+            try {
+                serverSocket = serverChannel.socket();
+            } finally {
+                if (serverSocket != null)
+                    closeables.add(serverSocket);
+            }
+
+
            serverSocket.setReuseAddress(true);

            serverSocket.bind(details.address());
@@ -661,6 +695,18 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
        }
    }

+    private ServerSocketChannel openServerSocketChannel() throws IOException {
+        ServerSocketChannel result = null;
+
+        try {
+            result = ServerSocketChannel.open();
+        } finally {
+            if (result != null)
+                closeables.add(result);
+        }
+        return result;
+    }
+
    private class ClientConnector extends AbstractConnector {

        private final Details details;
@@ -682,7 +728,9 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
        @Override
        SelectableChannel doConnect() throws IOException, InterruptedException {
            boolean success = false;
-            final SocketChannel socketChannel = SocketChannel.open();
+
+            final SocketChannel socketChannel = openSocketChannel();
+
            try {
                socketChannel.configureBlocking(false);
                socketChannel.socket().setReuseAddress(true);
@@ -737,6 +785,8 @@ class TcpReplicator extends AbstractChannelReplicator implements Closeable {
                }
            }
        }
+
+
    }

    /**