Use CHMV8 fork until we target JDK8, close AsyncHttpClient#886

Author: slandelle

providers/netty3/src/main/java/org/asynchttpclient/providers/netty/channel/pool/DefaultChannelPool.java

@@ -19,13 +19,13 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.asynchttpclient.AsyncHttpClientConfig;
 import org.asynchttpclient.providers.netty.channel.Channels;
+import org.asynchttpclient.providers.netty.chmv8.ConcurrentHashMapV8;
 import org.asynchttpclient.providers.netty.commons.channel.pool.ChannelPoolPartitionSelector;
 import org.asynchttpclient.providers.netty.future.NettyResponseFuture;
 import org.jboss.netty.channel.Channel;
@@ -43,8 +43,15 @@ public final class DefaultChannelPool implements ChannelPool {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DefaultChannelPool.class);
 
-    private final ConcurrentHashMap<String, ConcurrentLinkedQueue<IdleChannel>> partitions = new ConcurrentHashMap<>();
-    private final ConcurrentHashMap<Integer, ChannelCreation> channelId2Creation = new ConcurrentHashMap<>();
+    private static final ConcurrentHashMapV8.Fun<String, ConcurrentLinkedQueue<IdleChannel>> PARTITION_COMPUTER = new ConcurrentHashMapV8.Fun<String, ConcurrentLinkedQueue<IdleChannel>>() {
+        @Override
+        public ConcurrentLinkedQueue<IdleChannel> apply(String partitionId) {
+            return new ConcurrentLinkedQueue<>();
+        }
+    };
+
+    private final ConcurrentHashMapV8<String, ConcurrentLinkedQueue<IdleChannel>> partitions = new ConcurrentHashMapV8<>();
+    private final ConcurrentHashMapV8<Integer, ChannelCreation> channelId2Creation = new ConcurrentHashMapV8<>();
     private final AtomicBoolean isClosed = new AtomicBoolean(false);
     private final Timer nettyTimer;
     private final boolean sslConnectionPoolEnabled;
@@ -227,18 +234,6 @@ public void run(Timeout timeout) throws Exception {
         }
     }
 
-    private ConcurrentLinkedQueue<IdleChannel> getPartition(String partitionId) {
-        ConcurrentLinkedQueue<IdleChannel> partition = partitions.get(partitionId);
-        if (partition == null) {
-            // lazy init partition
-            ConcurrentLinkedQueue<IdleChannel> newPartition = new ConcurrentLinkedQueue<>();
-            partition = partitions.putIfAbsent(partitionId, newPartition);
-            if (partition == null)
-                partition = newPartition;
-        }
-        return partition;
-    }
-    
     public boolean offer(Channel channel, String partition) {
         if (isClosed.get() || (!sslConnectionPoolEnabled && channel.getPipeline().get(SslHandler.class) != null))
             return false;
@@ -248,7 +243,7 @@ public boolean offer(Channel channel, String partition) {
         if (isTTLExpired(channel, now))
             return false;
 
-        boolean added = getPartition(partition).add(new IdleChannel(channel, now));
+        boolean added = partitions.computeIfAbsent(partition, PARTITION_COMPUTER).add(new IdleChannel(channel, now));
         if (added)
             channelId2Creation.putIfAbsent(channel.getId(), new ChannelCreation(now, partition));
 

providers/netty3/src/main/java/org/asynchttpclient/providers/netty/chmv8/ConcurrentHashMapV8.java

@@ -0,0 +1,140 @@
+/*
+ * 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.
+ */
+
+/*
+ * Written by Doug Lea with assistance from members of JCP JSR-166
+ * Expert Group and released to the public domain, as explained at
+ * http://creativecommons.org/publicdomain/zero/1.0/
+ */
+
+package org.asynchttpclient.providers.netty.chmv8;
+
+
+/**
+ * A thread managed by a {@link ForkJoinPool}, which executes
+ * {@link ForkJoinTask}s.
+ * This class is subclassable solely for the sake of adding
+ * functionality -- there are no overridable methods dealing with
+ * scheduling or execution.  However, you can override initialization
+ * and termination methods surrounding the main task processing loop.
+ * If you do create such a subclass, you will also need to supply a
+ * custom {@link ForkJoinPool.ForkJoinWorkerThreadFactory} to
+ * {@linkplain ForkJoinPool#ForkJoinPool use it} in a {@code ForkJoinPool}.
+ *
+ * @since 1.7
+ * @author Doug Lea
+ */
+@SuppressWarnings("all")
+public class ForkJoinWorkerThread extends Thread {
+    /*
+     * ForkJoinWorkerThreads are managed by ForkJoinPools and perform
+     * ForkJoinTasks. For explanation, see the internal documentation
+     * of class ForkJoinPool.
+     *
+     * This class just maintains links to its pool and WorkQueue.  The
+     * pool field is set immediately upon construction, but the
+     * workQueue field is not set until a call to registerWorker
+     * completes. This leads to a visibility race, that is tolerated
+     * by requiring that the workQueue field is only accessed by the
+     * owning thread.
+     */
+
+    final ForkJoinPool pool;                // the pool this thread works in
+    final ForkJoinPool.WorkQueue workQueue; // work-stealing mechanics
+
+    /**
+     * Creates a ForkJoinWorkerThread operating in the given pool.
+     *
+     * @param pool the pool this thread works in
+     * @throws NullPointerException if pool is null
+     */
+    protected ForkJoinWorkerThread(ForkJoinPool pool) {
+        // Use a placeholder until a useful name can be set in registerWorker
+        super("aForkJoinWorkerThread");
+        this.pool = pool;
+        this.workQueue = pool.registerWorker(this);
+    }
+
+    /**
+     * Returns the pool hosting this thread.
+     *
+     * @return the pool
+     */
+    public ForkJoinPool getPool() {
+        return pool;
+    }
+
+    /**
+     * Returns the unique index number of this thread in its pool.
+     * The returned value ranges from zero to the maximum number of
+     * threads (minus one) that may exist in the pool, and does not
+     * change during the lifetime of the thread.  This method may be
+     * useful for applications that track status or collect results
+     * per-worker-thread rather than per-task.
+     *
+     * @return the index number
+     */
+    public int getPoolIndex() {
+        return workQueue.poolIndex >>> 1; // ignore odd/even tag bit
+    }
+
+    /**
+     * Initializes internal state after construction but before
+     * processing any tasks. If you override this method, you must
+     * invoke {@code super.onStart()} at the beginning of the method.
+     * Initialization requires care: Most fields must have legal
+     * default values, to ensure that attempted accesses from other
+     * threads work correctly even before this thread starts
+     * processing tasks.
+     */
+    protected void onStart() {
+    }
+
+    /**
+     * Performs cleanup associated with termination of this worker
+     * thread.  If you override this method, you must invoke
+     * {@code super.onTermination} at the end of the overridden method.
+     *
+     * @param exception the exception causing this thread to abort due
+     * to an unrecoverable error, or {@code null} if completed normally
+     */
+    protected void onTermination(Throwable exception) {
+    }
+
+    /**
+     * This method is required to be public, but should never be
+     * called explicitly. It performs the main run loop to execute
+     * {@link ForkJoinTask}s.
+     */
+    public void run() {
+        Throwable exception = null;
+        try {
+            onStart();
+            pool.runWorker(workQueue);
+        } catch (Throwable ex) {
+            exception = ex;
+        } finally {
+            try {
+                onTermination(exception);
+            } catch (Throwable ex) {
+                if (exception == null)
+                    exception = ex;
+            } finally {
+                pool.deregisterWorker(this, exception);
+            }
+        }
+    }
+}