Repository: netty/netty
Issue: [ForkJoinPool] Thread Local Leak detected
Labels: defect

The Atmosphere Framework bundles the [ForkJoinPool](https://github.com/netty/netty/blob/master/common/src/main/java/io/netty/util/internal/chmv8/ForkJoinPool.java) implementation of Netty. It seems there is a Thread Local leak as reported [here](https://github.com/Atmosphere/atmosphere/issues/2048)

``` java
SEVERE: The web application [/atmosphere-chat] created a ThreadLocal with key of type [java.lang.ThreadLocal] (value [java.lang.ThreadLocal@41a80e5a]) and a value of type [org.atmosphere.util.chmv8.ForkJoinPool.Submitter]
 (value [org.atmosphere.util.chmv8.ForkJoinPool$Submitter@450ae3fb]) but failed to remove it when the web application was stopped. This is very likely to create a memory leak.
```

When using the JDK7/8 implementation the leak doesn't occurs (no error message) so I suspect the current implementation leaks.


Issue comments:
Note that those `Submitter`s are just `int` wrappers, so I think the leak is more of an annoying warning than a severe issue.

Also the Netty `ForkJoinPool` is just a copy of the OpenJDK FJP. So maybe a simple up date to the latest OpenJDK version would resolve this.

@buchgr I wonder if it's actually a fork of Doug Lea's initial work for JSR166-e, before it was merged into OpenJDK and modified again.

Netty's version has this [ThreadLocal<Subscriber>](https://github.com/netty/netty/blob/netty-4.0.31.Final/common/src/main/java/io/netty/util/internal/chmv8/ForkJoinPool.java#L1082) than [OpenJDK hasn't](http://hg.openjdk.java.net/jdk8/jdk8/jdk/file/687fd7c7986d/src/share/classes/java/util/concurrent/ForkJoinPool.java#l1098).

@slandelle good point. I am not exactly sure. Just saying that updating it might resolve such things.

@buchgr I'm not sure you guys will want to fork OpenJDK as it's GPLv2. You might want to use [Doug Lea's JSR166](http://g.oswego.edu/dl/concurrency-interest/) work that's Creative Commons - public domain.

But I agree, upgrading should fix the issue as Tomcat doesn't report it against regular CHMv8.

@slandelle - Thanks for the heads up!

@normanmaurer - I'll assign to you for now...pending discussion about 5.0.

Fixed by https://github.com/netty/netty/pull/5270


Pull request review comments:

Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/internal/chmv8/ForkJoinPool.java b/common/src/main/java/io/netty/util/internal/chmv8/ForkJoinPool.java
index bfe4e9d269d..24bded99e5b 100644
--- a/common/src/main/java/io/netty/util/internal/chmv8/ForkJoinPool.java
+++ b/common/src/main/java/io/netty/util/internal/chmv8/ForkJoinPool.java
@@ -1078,8 +1078,13 @@ final boolean isApparentlyUnblocked() {
      * to avoid contention in one pool is likely to hold for others.
      * Lazily initialized on first submission (but null-checked
      * in other contexts to avoid unnecessary initialization).
+     *
+     * Note: this was changed to fix https://github.com/netty/netty/issues/4211
+     * Instead of using "ThreadLocal<Submitter>" like jsr166e, just use "ThreadLocal<int[]>" to
+     * avoid leaking the Submitter's class loader. Here "int[]" is just an array with exactly one
+     * int.
      */
-    static final ThreadLocal<Submitter> submitters;
+    static final ThreadLocal<int[]> submitters;
 
     /**
      * Creates a new ForkJoinWorkerThread. This factory is used unless
@@ -1478,10 +1483,10 @@ final void deregisterWorker(ForkJoinWorkerThread wt, Throwable ex) {
      * randomly modified upon collisions using xorshifts, which
      * requires a non-zero seed.
      */
-    static final class Submitter {
-        int seed;
-        Submitter(int s) { seed = s; }
-    }
+    //static final class Submitter {
+    //    int seed;
+    //    Submitter(int s) { seed = s; }
+    //}
 
     /**
      * Unless shutting down, adds the given task to a submission queue
@@ -1492,12 +1497,12 @@ static final class Submitter {
      * @param task the task. Caller must ensure non-null.
      */
     final void externalPush(ForkJoinTask<?> task) {
-        Submitter z = submitters.get();
+        int[] z = submitters.get();
         WorkQueue q; int r, m, s, n, am; ForkJoinTask<?>[] a;
         int ps = plock;
         WorkQueue[] ws = workQueues;
         if (z != null && ps > 0 && ws != null && (m = (ws.length - 1)) >= 0 &&
-                (q = ws[m & (r = z.seed) & SQMASK]) != null && r != 0 &&
+                (q = ws[m & (r = z[0]) & SQMASK]) != null && r != 0 &&
                 U.compareAndSwapInt(q, QLOCK, 0, 1)) { // lock
             if ((a = q.array) != null &&
                     (am = a.length - 1) > (n = (s = q.top) - q.base)) {
@@ -1533,18 +1538,18 @@ final void externalPush(ForkJoinTask<?> task) {
      */
     private void fullExternalPush(ForkJoinTask<?> task) {
         int r = 0; // random index seed
-        for (Submitter z = submitters.get();;) {
+        for (int[] z = submitters.get();;) {
             WorkQueue[] ws; WorkQueue q; int ps, m, k;
             if (z == null) {
                 if (U.compareAndSwapInt(this, INDEXSEED, r = indexSeed,
                         r += SEED_INCREMENT) && r != 0)
-                    submitters.set(z = new Submitter(r));
+                    submitters.set(z = new int[]{ r });
             }
             else if (r == 0) {                  // move to a different index
-                r = z.seed;
+                r = z[0];
                 r ^= r << 13;                   // same xorshift as WorkQueues
                 r ^= r >>> 17;
-                z.seed = r ^= (r << 5);
+                z[0] = r ^= (r << 5);
             }
             if ((ps = plock) < 0)
                 throw new RejectedExecutionException();
@@ -2313,12 +2318,12 @@ private boolean tryTerminate(boolean now, boolean enable) {
      * least one task.
      */
     static WorkQueue commonSubmitterQueue() {
-        Submitter z; ForkJoinPool p; WorkQueue[] ws; int m, r;
+        int[] z; ForkJoinPool p; WorkQueue[] ws; int m, r;
         return ((z = submitters.get()) != null &&
                 (p = common) != null &&
                 (ws = p.workQueues) != null &&
                 (m = ws.length - 1) >= 0) ?
-                ws[m & z.seed & SQMASK] : null;
+                ws[m & z[0] & SQMASK] : null;
     }
 
     /**
@@ -2326,11 +2331,11 @@ static WorkQueue commonSubmitterQueue() {
      */
     final boolean tryExternalUnpush(ForkJoinTask<?> task) {
         WorkQueue joiner; ForkJoinTask<?>[] a; int m, s;
-        Submitter z = submitters.get();
+        int[] z = submitters.get();
         WorkQueue[] ws = workQueues;
         boolean popped = false;
         if (z != null && ws != null && (m = ws.length - 1) >= 0 &&
-                (joiner = ws[z.seed & m & SQMASK]) != null &&
+                (joiner = ws[z[0] & m & SQMASK]) != null &&
                 joiner.base != (s = joiner.top) &&
                 (a = joiner.array) != null) {
             long j = (((a.length - 1) & (s - 1)) << ASHIFT) + ABASE;
@@ -2349,11 +2354,11 @@ final boolean tryExternalUnpush(ForkJoinTask<?> task) {
 
     final int externalHelpComplete(CountedCompleter<?> task) {
         WorkQueue joiner; int m, j;
-        Submitter z = submitters.get();
+        int[] z = submitters.get();
         WorkQueue[] ws = workQueues;
         int s = 0;
         if (z != null && ws != null && (m = ws.length - 1) >= 0 &&
-                (joiner = ws[(j = z.seed) & m & SQMASK]) != null && task != null) {
+                (joiner = ws[(j = z[0]) & m & SQMASK]) != null && task != null) {
             int scans = m + m + 1;
             long c = 0L;             // for stability check
             j |= 1;                  // poll odd queues
@@ -3279,7 +3284,7 @@ protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
             throw new Error(e);
         }
 
-        submitters = new ThreadLocal<Submitter>();
+        submitters = new ThreadLocal<int[]>();
         defaultForkJoinWorkerThreadFactory =
                 new DefaultForkJoinWorkerThreadFactory();
         modifyThreadPermission = new RuntimePermission("modifyThread");


Test patch: