Complete Servlet 3.1 non-blocking support

git-svn-id: https://svn.apache.org/repos/asf/tomcat/trunk@1705358 13f79535-47bb-0310-9956-ffa450edef68
chaofz · Sep 25, 2015 · 71765f1 · 71765f1
1 parent 9b5d05e
commit 71765f1
Show file tree

Hide file tree

Showing 5 changed files with 46 additions and 8 deletions.
diff --git a/java/org/apache/coyote/http2/Http2Parser.java b/java/org/apache/coyote/http2/Http2Parser.java
@@ -163,7 +163,7 @@ private void readDataFrame(int streamId, int flags, int payloadSize)
                     Integer.toString(streamId), Integer.toString(dataLength), padding));
         }
 
-        ByteBuffer dest = output.getInputByteBuffer(streamId, dataLength);
+        ByteBuffer dest = output.startRequestBodyFrame(streamId, payloadSize);
         if (dest == null) {
             swallow(streamId, dataLength, false);
             // Process padding before sending any notifications in case padding
@@ -185,7 +185,7 @@ private void readDataFrame(int streamId, int flags, int payloadSize)
                 if (endOfStream) {
                     output.receiveEndOfStream(streamId);
                 }
-                dest.notifyAll();
+                output.endRequestBodyFrame(streamId);
             }
         }
         if (padLength > 0) {
@@ -586,7 +586,8 @@ static interface Output {
         HpackDecoder getHpackDecoder();
 
         // Data frames
-        ByteBuffer getInputByteBuffer(int streamId, int payloadSize) throws Http2Exception;
+        ByteBuffer startRequestBodyFrame(int streamId, int payloadSize) throws Http2Exception;
+        void endRequestBodyFrame(int streamId) throws Http2Exception;
         void receiveEndOfStream(int streamId) throws ConnectionException;
         void swallowedPadding(int streamId, int paddingLength) throws ConnectionException, IOException;
 

diff --git a/java/org/apache/coyote/http2/Http2UpgradeHandler.java b/java/org/apache/coyote/http2/Http2UpgradeHandler.java
@@ -953,13 +953,21 @@ public HpackDecoder getHpackDecoder() {
 
 
     @Override
-    public ByteBuffer getInputByteBuffer(int streamId, int payloadSize) throws Http2Exception {
+    public ByteBuffer startRequestBodyFrame(int streamId, int payloadSize) throws Http2Exception {
         Stream stream = getStream(streamId, true);
         stream.checkState(FrameType.DATA);
         return stream.getInputByteBuffer();
     }
 
 
+
+    @Override
+    public void endRequestBodyFrame(int streamId) throws Http2Exception {
+        Stream stream = getStream(streamId, true);
+        stream.getInputBuffer().onDataAvailable();
+    }
+
+
     @Override
     public void receiveEndOfStream(int streamId) throws ConnectionException {
         Stream stream = getStream(streamId, connectionState.get().isNewStreamAllowed());

diff --git a/java/org/apache/coyote/http2/Stream.java b/java/org/apache/coyote/http2/Stream.java
@@ -21,6 +21,7 @@
 import java.util.Iterator;
 
 import org.apache.coyote.ActionCode;
+import org.apache.coyote.ContainerThreadMarker;
 import org.apache.coyote.InputBuffer;
 import org.apache.coyote.OutputBuffer;
 import org.apache.coyote.Request;
@@ -537,11 +538,31 @@ boolean isReady() {
         }
 
 
-        synchronized boolean isRegisteredForRead() {
+        synchronized boolean isRequestBodyFullyRead() {
+            return inBuffer.position() == 0 && isInputFinished();
+        }
+
+
+        synchronized int available() {
+            return inBuffer.position();
+        }
+
+
+        /*
+         * Called after placing some data in the inBuffer.
+         */
+        synchronized boolean onDataAvailable() {
             if (readInterest) {
                 readInterest = false;
+                coyoteRequest.action(ActionCode.DISPATCH_READ, null);
+                if (!ContainerThreadMarker.isContainerThread()) {
+                    coyoteRequest.action(ActionCode.DISPATCH_EXECUTE, null);
+                }
                 return true;
             } else {
+                synchronized (inBuffer) {
+                    inBuffer.notifyAll();
+                }
                 return false;
             }
         }

diff --git a/java/org/apache/coyote/http2/StreamProcessor.java b/java/org/apache/coyote/http2/StreamProcessor.java
@@ -154,6 +154,9 @@ public void action(ActionCode actionCode, Object param) {
             ((AtomicBoolean) param).set(getErrorState().isError());
             break;
         }
+        case AVAILABLE: {
+            request.setAvailable(stream.getInputBuffer().available());
+        }
 
         // Request attribute support
         case REQ_HOST_ADDR_ATTRIBUTE: {
@@ -295,7 +298,7 @@ public void action(ActionCode actionCode, Object param) {
         // Servlet 3.1 non-blocking I/O
         case REQUEST_BODY_FULLY_READ: {
             AtomicBoolean result = (AtomicBoolean) param;
-            result.set(stream.isInputFinished());
+            result.set(stream.getInputBuffer().isRequestBodyFullyRead());
             break;
         }
         case NB_READ_INTEREST: {
@@ -328,7 +331,6 @@ public void action(ActionCode actionCode, Object param) {
 
         // Unimplemented / to review
         case ACK:
-        case AVAILABLE:
         case CLOSE_NOW:
         case DISABLE_SWALLOW_INPUT:
         case END_REQUEST:

diff --git a/test/org/apache/coyote/http2/Http2TestBase.java b/test/org/apache/coyote/http2/Http2TestBase.java
@@ -718,13 +718,19 @@ public HpackDecoder getHpackDecoder() {
 
 
         @Override
-        public ByteBuffer getInputByteBuffer(int streamId, int payloadSize) {
+        public ByteBuffer startRequestBodyFrame(int streamId, int payloadSize) {
             lastStreamId = Integer.toString(streamId);
             trace.append(lastStreamId + "-Body-" + payloadSize + "\n");
             return null;
         }
 
 
+        @Override
+        public void endRequestBodyFrame(int streamId) throws Http2Exception {
+            // NO-OP
+        }
+
+
         @Override
         public void receiveEndOfStream(int streamId) {
             lastStreamId = Integer.toString(streamId);