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();
 }