Add assert to ensure we not create an endless loop and fix unsafe publication

Motivation:

[#6153] reports an endless loop that existed in the Recycler, while this was fixed adding a few asserts to ensure this remains fixed is a good thing. Beside this we also should ensure this can not escape the constructor to avoid unsafe publication.

Modifications:

- Add asserts
- Fix unsafe publication

Result:

More correct code.

common/src/main/java/io/netty/util/Recycler.java

@@ -249,10 +249,6 @@ public abstract class Recycler<T> {
         private WeakOrderQueue(Stack<?> stack, Thread thread) {
             head = tail = new Link();
             owner = new WeakReference<Thread>(thread);
-            synchronized (stack) {
-                next = stack.head;
-                stack.head = this;
-            }
 
             // Its important that we not store the Stack itself in the WeakOrderQueue as the Stack also is used in
             // the WeakHashMap as key. So just store the enclosed AtomicInteger which should allow to have the
@@ -260,13 +256,26 @@ public abstract class Recycler<T> {
             availableSharedCapacity = stack.availableSharedCapacity;
         }
 
+        static WeakOrderQueue newQueue(Stack<?> stack, Thread thread) {
+            WeakOrderQueue queue = new WeakOrderQueue(stack, thread);
+            // Done outside of the constructor to ensure WeakOrderQueue.this does not escape the constructor and so
+            // may be accessed while its still constructed.
+            stack.setHead(queue);
+            return queue;
+        }
+
+        private void setNext(WeakOrderQueue next) {
+            assert next != this;
+            this.next = next;
+        }
+
         /**
          * Allocate a new {@link WeakOrderQueue} or return {@code null} if not possible.
          */
         static WeakOrderQueue allocate(Stack<?> stack, Thread thread) {
             // We allocated a Link so reserve the space
             return reserveSpace(stack.availableSharedCapacity, LINK_CAPACITY)
-                    ? new WeakOrderQueue(stack, thread) : null;
+                    ? WeakOrderQueue.newQueue(stack, thread) : null;
         }
 
         private static boolean reserveSpace(AtomicInteger availableSharedCapacity, int space) {
@@ -430,6 +439,12 @@ public abstract class Recycler<T> {
             this.maxDelayedQueues = maxDelayedQueues;
         }
 
+        // Marked as synchronized to ensure this is serialized.
+        synchronized void setHead(WeakOrderQueue queue) {
+            queue.setNext(head);
+            head = queue;
+        }
+
         int increaseCapacity(int expectedCapacity) {
             int newCapacity = elements.length;
             int maxCapacity = this.maxCapacity;
@@ -497,7 +512,6 @@ public abstract class Recycler<T> {
                     success = true;
                     break;
                 }
-
                 WeakOrderQueue next = cursor.next;
                 if (cursor.owner.get() == null) {
                     // If the thread associated with the queue is gone, unlink it, after
@@ -512,8 +526,9 @@ public abstract class Recycler<T> {
                             }
                         }
                     }
+
                     if (prev != null) {
-                        prev.next = next;
+                        prev.setNext(next);
                     }
                 } else {
                     prev = cursor;