Enforce ratioMask also for WeakOrderQueue (#9727)

Motivation:

At the moment we only enfore ratioMask for the Stack which means that we only guard against recycle burts when recycled from the same Thread. We should also enforce the ratioMask in the WeakOrderQueue so we also guard against the bursts when recycle from other threads.

Modifications:

- Keep counter in WeakOrderQueue to enforce ratioMask as well
- Adjust unit test

Result:

Better guard against recycle bursts which could pollute the heap unnecessary.

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

@@ -303,10 +303,13 @@ public abstract class Recycler<T> {
         // pointer to another queue of delayed items for the same stack
         private WeakOrderQueue next;
         private final int id = ID_GENERATOR.getAndIncrement();
+        private final int ratioMask;
+        private int handleRecycleCount;
 
         private WeakOrderQueue() {
             super(null);
             head = new Head(null);
+            ratioMask = 0;
         }
 
         private WeakOrderQueue(Stack<?> stack, Thread thread) {
@@ -318,6 +321,7 @@ public abstract class Recycler<T> {
             // Stack itself GCed.
             head = new Head(stack.availableSharedCapacity);
             head.link = tail;
+            ratioMask = stack.ratioMask;
         }
 
         private static WeakOrderQueue newQueue(Stack<?> stack, Thread thread) {
@@ -355,6 +359,14 @@ public abstract class Recycler<T> {
         void add(DefaultHandle<?> handle) {
             handle.lastRecycledId = id;
 
+            // While we also enforce the recycling ratio one we transfer objects from the WeakOrderQueue to the Stack
+            // we better should enforce it as well early. Missing to do so may let the WeakOrderQueue grow very fast
+            // without control if the Stack
+            if ((++handleRecycleCount & ratioMask) != 0) {
+                // Drop the item to prevent recycling to aggressive.
+                return;
+            }
+
             Link tail = this.tail;
             int writeIndex;
             if ((writeIndex = tail.get()) == LINK_CAPACITY) {
@@ -637,6 +649,7 @@ public abstract class Recycler<T> {
                 // We don't support recycling across threads and should just drop the item on the floor.
                 return;
             }
+
             // we don't want to have a ref to the queue as the value in our weak map
             // so we null it out; to ensure there are no races with restoring it later
             // we impose a memory ordering here (no-op on x86)

common/src/test/java/io/netty/util/RecyclerTest.java

@@ -163,6 +163,7 @@ public class RecyclerTest {
 
         final HandledObject o = recycler.get();
         final HandledObject o2 = recycler.get();
+
         final Thread thread = new Thread() {
             @Override
             public void run() {
@@ -173,8 +174,9 @@ public class RecyclerTest {
         thread.start();
         thread.join();
 
-        assertSame(recycler.get(), o);
-        assertNotSame(recycler.get(), o2);
+        // As we use a ratioMask of 2 we should see o2 as the first object that could recycled from a different thread.
+        assertSame(recycler.get(), o2);
+        assertNotSame(recycler.get(), o);
     }
 
     @Test