Use fast thread local and fix GC nepotism (Fixes #4749) (#4750)

* Use fast thread local and fix GC nepotism (Fixes #4749)
* Replace order of reentrant task's merge

src/main/java/io/vertx/core/net/impl/pool/CombinerExecutor.java

@@ -10,6 +10,7 @@
  */
 package io.vertx.core.net.impl.pool;
 
+import io.netty.util.concurrent.FastThreadLocal;
 import io.netty.util.internal.PlatformDependent;
 
 import java.util.Queue;
@@ -27,7 +28,12 @@ public class CombinerExecutor<S> implements Executor<S> {
   private final Queue<Action<S>> q = PlatformDependent.newMpscQueue();
   private final AtomicInteger s = new AtomicInteger();
   private final S state;
-  private final ThreadLocal<Task> current = new ThreadLocal<>();
+
+  protected final class InProgressTail {
+    Task task;
+  }
+
+  private final FastThreadLocal<InProgressTail> current = new FastThreadLocal<>();
 
   public CombinerExecutor(S state) {
     this.state = state;
@@ -40,50 +46,46 @@ public void submit(Action<S> action) {
       return;
     }
     Task head = null;
+    Task tail = null;
     do {
       try {
-        head = pollAndExecute(head);
+        final Action<S> a = q.poll();
+        if (a == null) {
+          break;
+        }
+        Task task = a.execute(state);
+        if (task != null) {
+          if (head == null) {
+            assert tail == null;
+            tail = task;
+            head = task;
+          } else {
+            tail = tail.next(task);
+          }
+        }
       } finally {
         s.set(0);
       }
     } while (!q.isEmpty() && s.compareAndSet(0, 1));
     if (head != null) {
-      Task inProgress = current.get();
+      InProgressTail inProgress = current.get();
       if (inProgress == null) {
-        current.set(head);
+        inProgress = new InProgressTail();
+        current.set(inProgress);
+        inProgress.task = tail;
         try {
-          while (head != null) {
-            head.run();
-            head = head.next;
-          }
+          // from now one cannot trust tail anymore
+          head.runNextTasks();
         } finally {
           current.remove();
         }
       } else {
-        merge(inProgress, head);
-      }
-    }
-  }
+        assert inProgress.task != null;
+        Task oldNextTail = inProgress.task.replaceNext(head);
+        assert oldNextTail == null;
+        inProgress.task = tail;
 
-  private Task pollAndExecute(Task head) {
-    Action<S> action;
-    while ((action = q.poll()) != null) {
-      Task task = action.execute(state);
-      if (task != null) {
-        if (head == null) {
-          head = task;
-        } else {
-          merge(head, task);
-        }
       }
     }
-    return head;
-  }
-
-  private static void merge(Task head, Task tail) {
-    Task tmp = tail.prev;
-    tail.prev = head.prev;
-    head.prev.next = tail;
-    head.prev = tmp;
   }
 }

src/main/java/io/vertx/core/net/impl/pool/Task.java

@@ -12,9 +12,29 @@
 
 public abstract class Task {
 
-  Task prev = this;
-  Task next;
+  private Task next;
 
-  public abstract void run();
+  public Task replaceNext(Task next) {
+    Task oldNext = this.next;
+    this.next = next;
+    return oldNext;
+  }
+
+  public Task next(Task next) {
+    this.next = next;
+    return next;
+  }
 
+  protected final void runNextTasks() {
+    Task task = this;
+    while (task != null) {
+      task.run();
+      final Task next = task.next;
+      // help GC :P
+      task.next = null;
+      task = next;
+    }
+  }
+
+  public abstract void run();
 }