Repository: netty/netty
Issue: Livelock issue in MpscLinkedQueue.java
Labels: defect

Hi,

We've been running netty 4.0.27 in production for ~1 week, and recently we started noticing cases where calls to `SingleThreadEventExecutor.pendingTasks()` were never completing. We traced the issue down to `MpscLinkedQueue.peekNode()`, which appears to be just spinning. 

Reading the contract at the top of the file seems to indicate that calling `MpscLinkedQueue.size()` should be ok from multiple threads, but perhaps I am reading it incorrectly?

Unfortunately I don't have a good way of reproducing this, it only showed up in production. For now we've removed the code that was calling `.size()`, but if anyone has suggestions on a way to try reproducing the bug (or suggestions as to how we're violating the queue contract), I'd be happy to try.


Issue comments:
@gsoltis so it only happens when you call size() ?

So far, yes, that's the only place I've seen it. We have each event loop hooked up to a stats reporting module that reports the size of the task queue every few seconds. We noticed it because the stats stopped being reported, and it turns out the reporting thread was blocked on the call to `.size()`, which in turn was spinning on `peekNode()`. 

Here's the last few lines of the stacktrace:

> "StatsDLogger" prio=10 tid=0x00007f8e08f83800 nid=0x5664 runnable [0x00007f78f105b000]
>   java.lang.Thread.State: RUNNABLE
>   at io.netty.util.internal.MpscLinkedQueue.peekNode(MpscLinkedQueue.java:109)
>   at io.netty.util.internal.MpscLinkedQueue.size(MpscLinkedQueue.java:166)
>   at 
> io.netty.util.concurrent.SingleThreadEventExecutor.pendingTasks(SingleThreadEventExecutor.java:287)

@gsoltis will check... Thanks for the report buddy!

No problem. Let me know if there's anything else I can provide. I realize
it's not a lot to go on :)

On Mon, Apr 20, 2015, 10:02 PM Norman Maurer notifications@github.com
wrote:

> @gsoltis https://github.com/gsoltis will check... Thanks for the report
> buddy!
> 
> —
> Reply to this email directly or view it on GitHub
> https://github.com/netty/netty/issues/3675#issuecomment-94639423.

@gsoltis what I can tell you is that size() is now allowed to be called by another thread then the EventLoop. Thinking about a proper fix.

@gsoltis can you please check if #3684 fixes the problem for you ? Thanks!

I'll give it a shot. It may be a few days before I can both get it in production and verify that it's working, since it took so long to show up previously.

Ok just let me know once you have results

> Am 21.04.2015 um 23:12 schrieb Greg Soltis notifications@github.com:
> 
> I'll give it a shot. It may be a few days before I can both get it in production and verify that it's working, since it took so long to show up previously.
> 
> —
> Reply to this email directly or view it on GitHub.

@gsoltis I think the change fixed it... Please re-open if you still see the issue.

Will do, thanks for your help!

On Tue, May 5, 2015 at 9:23 PM, Norman Maurer notifications@github.com
wrote:

> Closed #3675 https://github.com/netty/netty/issues/3675.
> 
> —
> Reply to this email directly or view it on GitHub
> https://github.com/netty/netty/issues/3675#event-297870903.


Pull request review comments:
Does the runtime type of `toList(..)` have to be `ArrayList<>`?  Can you just change the signature of  `toList(..)` to accept a `List<..>` since it is private?  That way we don't have to duplicate the `16 is the default size` logic here. 

@Scottmitch sorry but I not understand what you are proposing.. IF I understood you right you may want to have toList accept a List<...>, I just not see why this is better then what we have now. Can you explain ?

Where did `16` come from?  Is this a value we have explicitly decided is good...or are we just using it because it is also the default value of some common collection implementations in Java? If we have justification why this makes sense...then never mind me (but add a comment :) ).  Otherwise if we don't have any rational for the size and just choose it because that is what default implementations choose then lets just let the implementation choose the default (i.e. `new ArrayList<>()`) and change the parameter from a `int` to a `List<>`.

@Scottmitch The 16 came from the mighty @normanmaurer =P Joking aside I just thought it may not be a bad fit as queues often have some items in there. That said I not care too much about this method as we not use any of these methods that call it in our code-base. 

> we not use any of these methods that call it in our code-base.

Sounds like the mighty @normanmaurer is setting a trap for our users :)

honestly it is in the internal package... That said if you have a better idea shoot :+1: 

Agreed not a big deal...but why have magic numbers when you can not have them...

Change this:

``` java
private List<E> toList(int initialCapacity) {
..
}
@Override
public Object[] toArray() {
    return toList(16).toArray();
}
```

to this:

``` java
private List<E> toList(List<E> elements) {
..
}
private List<E> toList(int initialCapacity) {
  return toList(new ArrayList<E>(initialCapacity));
}
private List<E> toList() {
  return toList(new ArrayList<E>());
}
@Override
public Object[] toArray() {
  return toList().toArray();
}
```

then also search and replace the other magic `16` number floating around in [this file](https://github.com/netty/netty/pull/3684/files#diff-ac5cb182f74f909912b9cce8a20556aaR214)

This would be a candidate for https://github.com/netty/netty/pull/3684/files#r29030165

fixed


Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/internal/MpscLinkedQueue.java b/common/src/main/java/io/netty/util/internal/MpscLinkedQueue.java
index 4baa5ac0fb2..c1b931a5628 100644
--- a/common/src/main/java/io/netty/util/internal/MpscLinkedQueue.java
+++ b/common/src/main/java/io/netty/util/internal/MpscLinkedQueue.java
@@ -21,10 +21,10 @@
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
-import java.lang.reflect.Array;
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.Queue;
 
@@ -164,12 +164,20 @@ public E peek() {
     public int size() {
         int count = 0;
         MpscLinkedQueueNode<E> n = peekNode();
-        for (;;) {
-            if (n == null) {
+         for (;;) {
+            // If value == null it means that clearMaybe() was called on the MpscLinkedQueueNode.
+            if (n == null || n.value() == null) {
+                break;
+            }
+            MpscLinkedQueueNode<E> next = n.next();
+            if (n == next) {
+                break;
+            }
+            n = next;
+            if (++ count == Integer.MAX_VALUE) {
+                // Guard against overflow of integer.
                 break;
             }
-            count ++;
-            n = n.next();
         }
         return count;
     }
@@ -186,40 +194,26 @@ public boolean contains(Object o) {
             if (n == null) {
                 break;
             }
-            if (n.value() == o) {
+            E value = n.value();
+            // If value == null it means that clearMaybe() was called on the MpscLinkedQueueNode.
+            if (value == null) {
+                return false;
+            }
+            if (value == o) {
                 return true;
             }
-            n = n.next();
+            MpscLinkedQueueNode<E> next = n.next();
+            if (n == next) {
+                break;
+            }
+            n = next;
         }
         return false;
     }
 
     @Override
     public Iterator<E> iterator() {
-        return new Iterator<E>() {
-            private MpscLinkedQueueNode<E> node = peekNode();
-
-            @Override
-            public boolean hasNext() {
-                return node != null;
-            }
-
-            @Override
-            public E next() {
-                MpscLinkedQueueNode<E> node = this.node;
-                if (node == null) {
-                    throw new NoSuchElementException();
-                }
-                E value = node.value();
-                this.node = node.next();
-                return value;
-            }
-
-            @Override
-            public void remove() {
-                throw new UnsupportedOperationException();
-            }
-        };
+        return new ReadOnlyIterator<E>(toList().iterator());
     }
 
     @Override
@@ -248,53 +242,46 @@ public E element() {
         throw new NoSuchElementException();
     }
 
-    @Override
-    public Object[] toArray() {
-        final Object[] array = new Object[size()];
-        final Iterator<E> it = iterator();
-        for (int i = 0; i < array.length; i ++) {
-            if (it.hasNext()) {
-                array[i] = it.next();
-            } else {
-                return Arrays.copyOf(array, i);
+    private List<E> toList(int initialCapacity) {
+        return toList(new ArrayList<E>(initialCapacity));
+    }
+
+    private List<E> toList() {
+        return toList(new ArrayList<E>());
+    }
+
+    private List<E> toList(List<E> elements) {
+        MpscLinkedQueueNode<E> n = peekNode();
+        for (;;) {
+            if (n == null) {
+                break;
+            }
+            E value = n.value();
+            if (value == null) {
+                break;
+            }
+            if (!elements.add(value)) {
+                // Seems like there is no space left, break here.
+                break;
+            }
+            MpscLinkedQueueNode<E> next = n.next();
+            if (n == next) {
+                break;
             }
+            n = next;
         }
-        return array;
+        return elements;
+    }
+
+    @Override
+    public Object[] toArray() {
+        return toList().toArray();
     }
 
     @Override
     @SuppressWarnings("unchecked")
     public <T> T[] toArray(T[] a) {
-        final int size = size();
-        final T[] array;
-        if (a.length >= size) {
-            array = a;
-        } else {
-            array = (T[]) Array.newInstance(a.getClass().getComponentType(), size);
-        }
-
-        final Iterator<E> it = iterator();
-        for (int i = 0; i < array.length; i++) {
-            if (it.hasNext()) {
-                array[i] = (T) it.next();
-            } else {
-                if (a == array) {
-                    array[i] = null;
-                    return array;
-                }
-
-                if (a.length < i) {
-                    return Arrays.copyOf(array, i);
-                }
-
-                System.arraycopy(array, 0, a, 0, i);
-                if (a.length > i) {
-                    a[i] = null;
-                }
-                return a;
-            }
-        }
-        return array;
+        return toList(a.length).toArray(a);
     }
 
     @Override


Test patch: