simplified HttpInput lock design

jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java

@@ -63,13 +63,22 @@ public abstract class HttpInput<T> extends ServletInputStream implements Runnabl
     
     protected State _state = BLOCKING;
     private State _eof=null;
-    
-    
+    private final Object _lock;
+
     protected HttpInput()
     {
+        this(null);
     }
     
-    public abstract Object lock();
+    protected HttpInput(Object lock)
+    {
+        _lock=lock==null?this:lock;
+    }
+    
+    public final Object lock()
+    {
+        return _lock;
+    }
 
     public void recycle()
     {
@@ -431,7 +440,10 @@ public abstract class HttpInput<T> extends ServletInputStream implements Runnabl
 
     public void init(HttpChannelState state)
     {
-        _channelState=state;
+        synchronized (lock())
+        {
+            _channelState=state;
+        }
     }
 
 }

jetty-server/src/main/java/org/eclipse/jetty/server/HttpInputOverHTTP.java

@@ -53,7 +53,6 @@ class HttpInputOverHTTP extends HttpInput<ByteBuffer> implements Callback
         }
     }
 
-
     @Override
     protected void blockForContent() throws IOException
     {
@@ -74,12 +73,6 @@ class HttpInputOverHTTP extends HttpInput<ByteBuffer> implements Callback
     {
         return String.format("%s@%x",getClass().getSimpleName(),hashCode());
     }
-    
-    @Override
-    public Object lock()
-    {
-        return this;
-    }
 
     @Override
     protected ByteBuffer nextContent() throws IOException
@@ -120,7 +113,6 @@ class HttpInputOverHTTP extends HttpInput<ByteBuffer> implements Callback
                 }
                 return null;
             }
-            
         }
             
         return null; 

jetty-server/src/main/java/org/eclipse/jetty/server/QueuedHttpInput.java

@@ -42,16 +42,10 @@ public abstract class QueuedHttpInput<T> extends HttpInput<T>
 {
     private final static Logger LOG = Log.getLogger(QueuedHttpInput.class);
 
-    private final ArrayQueue<T> _inputQ = new ArrayQueue<>();
+    private final ArrayQueue<T> _inputQ = new ArrayQueue<>(lock());
     
     public QueuedHttpInput()
-    {
-    }
-    
-    public Object lock()
-    {
-        return _inputQ.lock();
-    }
+    {}
 
     public void recycle()
     {

jetty-util/src/main/java/org/eclipse/jetty/util/ArrayQueue.java

@@ -48,6 +48,12 @@ public class ArrayQueue<E> extends AbstractList<E> implements Queue<E>
     {
         this(DEFAULT_CAPACITY, -1);
     }
+    
+    /* ------------------------------------------------------------ */
+    public ArrayQueue(Object lock)
+    {
+        this(DEFAULT_CAPACITY, -1,lock);
+    }
 
     /* ------------------------------------------------------------ */
     public ArrayQueue(int capacity)