lock all HttpInput methods to prevents concurrent threads to work on the same ByteBuffer

Signed-off-by: Ludovic Orban <lorban@bitronix.be>

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

@@ -23,6 +23,7 @@ import org.eclipse.jetty.server.handler.ContextHandler;
 import org.eclipse.jetty.util.BufferUtil;
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.component.Destroyable;
+import org.eclipse.jetty.util.thread.AutoLock;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,14 +43,18 @@ public class HttpInput extends ServletInputStream implements Runnable
     private boolean _consumedEof;
     private ReadListener _readListener;
     private long _contentConsumed;
+    private final AutoLock _lock = new AutoLock();
 
     public HttpInput(HttpChannelState state)
+    {
+        try (AutoLock lock = _lock.lock())
         {
             _channelState = state;
             _asyncContentProducer = new AsyncContentProducer(state.getHttpChannel());
             _blockingContentProducer = new BlockingContentProducer(_asyncContentProducer);
             _contentProducer = _blockingContentProducer;
         }
+    }
 
     public void recycle()
     {
@@ -58,6 +63,8 @@ public class HttpInput extends ServletInputStream implements Runnable
     }
 
     public void reopen()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("reopen {}", this);
@@ -67,14 +74,18 @@ public class HttpInput extends ServletInputStream implements Runnable
             _readListener = null;
             _contentConsumed = 0;
         }
+    }
 
     /**
      * @return The current Interceptor, or null if none set
      */
     public Interceptor getInterceptor()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             return _contentProducer.getInterceptor();
         }
+    }
 
     /**
      * Set the interceptor.
@@ -82,11 +93,14 @@ public class HttpInput extends ServletInputStream implements Runnable
      * @param interceptor The interceptor to use.
      */
     public void setInterceptor(Interceptor interceptor)
+    {
+        try (AutoLock lock = _lock.lock())
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("setting interceptor to {} on {}", interceptor, this);
             _contentProducer.setInterceptor(interceptor);
         }
+    }
 
     /**
      * Set the {@link Interceptor}, chaining it to the existing one if
@@ -95,6 +109,8 @@ public class HttpInput extends ServletInputStream implements Runnable
      * @param interceptor the next {@link Interceptor} in a chain
      */
     public void addInterceptor(Interceptor interceptor)
+    {
+        try (AutoLock lock = _lock.lock())
         {
             Interceptor currentInterceptor = _contentProducer.getInterceptor();
             if (currentInterceptor == null)
@@ -111,8 +127,9 @@ public class HttpInput extends ServletInputStream implements Runnable
                 _contentProducer.setInterceptor(chainedInterceptor);
             }
         }
+    }
 
-    public int get(Content content, byte[] bytes, int offset, int length)
+    private int get(Content content, byte[] bytes, int offset, int length)
     {
         int consumed = content.get(bytes, offset, length);
         _contentConsumed += consumed;
@@ -120,16 +137,24 @@ public class HttpInput extends ServletInputStream implements Runnable
     }
 
     public long getContentConsumed()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             return _contentConsumed;
         }
+    }
 
     public long getContentReceived()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             return _contentProducer.getRawContentArrived();
         }
+    }
 
     public boolean consumeAll()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             IOException failure = new IOException("Unconsumed content");
             if (LOG.isDebugEnabled())
@@ -143,44 +168,59 @@ public class HttpInput extends ServletInputStream implements Runnable
 
             return false;
         }
+    }
 
     public boolean isError()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             boolean error = _contentProducer.isError();
             if (LOG.isDebugEnabled())
                 LOG.debug("isError={} {}", error, this);
             return error;
         }
+    }
 
     public boolean isAsync()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("isAsync read listener {} {}", _readListener, this);
             return _readListener != null;
         }
+    }
 
     /* ServletInputStream */
 
     @Override
     public boolean isFinished()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             boolean finished = _consumedEof;
             if (LOG.isDebugEnabled())
                 LOG.debug("isFinished={} {}", finished, this);
             return finished;
         }
+    }
 
     @Override
     public boolean isReady()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             boolean ready = _contentProducer.isReady();
             if (LOG.isDebugEnabled())
                 LOG.debug("isReady={} {}", ready, this);
             return ready;
         }
+    }
 
     @Override
     public void setReadListener(ReadListener readListener)
+    {
+        try (AutoLock lock = _lock.lock())
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("setting read listener to {} {}", readListener, this);
@@ -196,23 +236,32 @@ public class HttpInput extends ServletInputStream implements Runnable
             if (isReady() && _channelState.onReadEof()) // onReadEof b/c we want to transition from WAITING to WOKEN
                 scheduleReadListenerNotification(); // this is needed by AsyncServletIOTest.testStolenAsyncRead
         }
+    }
 
     public boolean onContentProducible()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             return _contentProducer.onContentProducible();
         }
+    }
 
     @Override
     public int read() throws IOException
+    {
+        try (AutoLock lock = _lock.lock())
         {
             int read = read(_oneByteBuffer, 0, 1);
             if (read == 0)
                 throw new IOException("unready read=0");
             return read < 0 ? -1 : _oneByteBuffer[0] & 0xFF;
         }
+    }
 
     @Override
     public int read(byte[] b, int off, int len) throws IOException
+    {
+        try (AutoLock lock = _lock.lock())
         {
             // Calculate minimum request rate for DoS protection
             _contentProducer.checkMinDataRate();
@@ -253,6 +302,7 @@ public class HttpInput extends ServletInputStream implements Runnable
 
             throw new AssertionError("no data, no error and not EOF");
         }
+    }
 
     private void scheduleReadListenerNotification()
     {
@@ -266,6 +316,8 @@ public class HttpInput extends ServletInputStream implements Runnable
      * @return true if the input contains content, false otherwise.
      */
     public boolean hasContent()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             // Do not call _contentProducer.available() as it calls HttpChannel.produceContent()
             // which is forbidden by this method's contract.
@@ -274,15 +326,19 @@ public class HttpInput extends ServletInputStream implements Runnable
                 LOG.debug("hasContent={} {}", hasContent, this);
             return hasContent;
         }
+    }
 
     @Override
     public int available()
+    {
+        try (AutoLock lock = _lock.lock())
         {
             int available = _contentProducer.available();
             if (LOG.isDebugEnabled())
                 LOG.debug("available={} {}", available, this);
             return available;
         }
+    }
 
     /* Runnable */
 
@@ -292,6 +348,10 @@ public class HttpInput extends ServletInputStream implements Runnable
      */
     @Override
     public void run()
+    {
+        Content content;
+        ReadListener readListener;
+        try (AutoLock lock = _lock.lock())
         {
             // Call isReady() to make sure that if not ready we register for fill interest.
             if (!_contentProducer.isReady())
@@ -300,12 +360,15 @@ public class HttpInput extends ServletInputStream implements Runnable
                     LOG.debug("running but not ready {}", this);
                 return;
             }
-        Content content = _contentProducer.nextContent();
+            content = _contentProducer.nextContent();
             if (LOG.isDebugEnabled())
                 LOG.debug("running on content {} {}", content, this);
 
+            readListener = this._readListener;
+        }
+
         // This check is needed when a request is started async but no read listener is registered.
-        if (_readListener == null)
+        if (readListener == null)
         {
             if (LOG.isDebugEnabled())
                 LOG.debug("running without a read listener {}", this);
@@ -322,7 +385,7 @@ public class HttpInput extends ServletInputStream implements Runnable
                     LOG.debug("running error={} {}", error, this);
                 // TODO is this necessary to add here?
                 _channelState.getHttpChannel().getResponse().getHttpFields().add(HttpConnection.CONNECTION_CLOSE);
-                _readListener.onError(error);
+                readListener.onError(error);
             }
             else if (content.isEof())
             {
@@ -330,13 +393,13 @@ public class HttpInput extends ServletInputStream implements Runnable
                 {
                     if (LOG.isDebugEnabled())
                         LOG.debug("running at EOF {}", this);
-                    _readListener.onAllDataRead();
+                    readListener.onAllDataRead();
                 }
                 catch (Throwable x)
                 {
                     if (LOG.isDebugEnabled())
                         LOG.debug("running failed onAllDataRead {}", this, x);
-                    _readListener.onError(x);
+                    readListener.onError(x);
                 }
             }
         }
@@ -346,13 +409,13 @@ public class HttpInput extends ServletInputStream implements Runnable
                 LOG.debug("running has content {}", this);
             try
             {
-                _readListener.onDataAvailable();
+                readListener.onDataAvailable();
             }
             catch (Throwable x)
             {
                 if (LOG.isDebugEnabled())
                     LOG.debug("running failed onDataAvailable {}", this, x);
-                _readListener.onError(x);
+                readListener.onError(x);
             }
         }
     }