Avoid register multiple cleaner task for same thread's FastThreadLocal index

Motivation:

Currently if user call set/remove/set/remove many times, it will create multiple cleaner task for same index. It may cause OOM due to long live thread will have more and more task in LIVE_SET.

Modification:

Add flag to avoid recreating tasks.

Result:
Only create 1 clean task. But use more space of indexedVariables.

common/src/main/java/io/netty/util/concurrent/FastThreadLocal.java

@@ -125,8 +125,11 @@ public class FastThreadLocal<V> {
 
     private final int index;
 
+    private final int cleanerFlagIndex;
+
     public FastThreadLocal() {
         index = InternalThreadLocalMap.nextVariableIndex();
+        cleanerFlagIndex = InternalThreadLocalMap.nextVariableIndex();
     }
 
     /**
@@ -147,7 +150,14 @@ public class FastThreadLocal<V> {
 
     private void registerCleaner(final InternalThreadLocalMap threadLocalMap) {
         Thread current = Thread.currentThread();
-        if (!FastThreadLocalThread.willCleanupFastThreadLocals(current)) {
+        if (FastThreadLocalThread.willCleanupFastThreadLocals(current) ||
+            threadLocalMap.indexedVariable(cleanerFlagIndex) != InternalThreadLocalMap.UNSET) {
+            return;
+        }
+        // removeIndexedVariable(cleanerFlagIndex) isn't necessary because the finally cleanup is tied to the lifetime
+        // of the thread, and this Object will be discarded if the associated thread is GCed.
+        threadLocalMap.setIndexedVariable(cleanerFlagIndex, Boolean.TRUE);
+
         // We will need to ensure we will trigger remove(InternalThreadLocalMap) so everything will be released
         // and FastThreadLocal.onRemoval(...) will be called.
         ObjectCleaner.register(current, new Runnable() {
@@ -160,7 +170,6 @@ public class FastThreadLocal<V> {
             }
         });
     }
-    }
 
     /**
      * Returns the current value for the specified thread local map.

common/src/main/java/io/netty/util/internal/ObjectCleaner.java

@@ -123,6 +123,10 @@ public final class ObjectCleaner {
         }
     }
 
+    public static int getLiveSetCount() {
+        return LIVE_SET.size();
+    }
+
     private ObjectCleaner() {
         // Only contains a static method.
     }

common/src/test/java/io/netty/util/concurrent/FastThreadLocalTest.java

@@ -16,6 +16,7 @@
 
 package io.netty.util.concurrent;
 
+import io.netty.util.internal.ObjectCleaner;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -24,7 +25,8 @@ import java.util.concurrent.atomic.AtomicReference;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.Matchers.nullValue;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 
 public class FastThreadLocalTest {
     @Before
@@ -45,7 +47,7 @@ public class FastThreadLocalTest {
 
         // Initialize a thread-local variable.
         assertThat(var.get(), is(nullValue()));
-        assertThat(FastThreadLocal.size(), is(1));
+        assertThat(FastThreadLocal.size(), is(2));
 
         // And then remove it.
         FastThreadLocal.removeAll();
@@ -76,6 +78,65 @@ public class FastThreadLocalTest {
         }
     }
 
+    @Test
+    public void testMultipleSetRemove() throws Exception {
+        final FastThreadLocal<String> threadLocal = new FastThreadLocal<String>();
+        final Runnable runnable = new Runnable() {
+            @Override
+            public void run() {
+                threadLocal.set("1");
+                threadLocal.remove();
+                threadLocal.set("2");
+                threadLocal.remove();
+            }
+        };
+
+        final int sizeWhenStart = ObjectCleaner.getLiveSetCount();
+        Thread thread = new Thread(runnable);
+        thread.start();
+        thread.join();
+
+        assertEquals(1, ObjectCleaner.getLiveSetCount() - sizeWhenStart);
+
+        Thread thread2 = new Thread(runnable);
+        thread2.start();
+        thread2.join();
+
+        assertEquals(2, ObjectCleaner.getLiveSetCount() - sizeWhenStart);
+    }
+
+    @Test
+    public void testMultipleSetRemove_multipleThreadLocal() throws Exception {
+        final FastThreadLocal<String> threadLocal = new FastThreadLocal<String>();
+        final FastThreadLocal<String> threadLocal2 = new FastThreadLocal<String>();
+        final Runnable runnable = new Runnable() {
+            @Override
+            public void run() {
+                threadLocal.set("1");
+                threadLocal.remove();
+                threadLocal.set("2");
+                threadLocal.remove();
+                threadLocal2.set("1");
+                threadLocal2.remove();
+                threadLocal2.set("2");
+                threadLocal2.remove();
+            }
+        };
+
+        final int sizeWhenStart = ObjectCleaner.getLiveSetCount();
+        Thread thread = new Thread(runnable);
+        thread.start();
+        thread.join();
+
+        assertEquals(2, ObjectCleaner.getLiveSetCount() - sizeWhenStart);
+
+        Thread thread2 = new Thread(runnable);
+        thread2.start();
+        thread2.join();
+
+        assertEquals(4, ObjectCleaner.getLiveSetCount() - sizeWhenStart);
+    }
+
     @Test(timeout = 4000)
     public void testOnRemoveCalledForFastThreadLocalGet() throws Exception {
         testOnRemoveCalled(true, true);