Move the additional dispatches required from the SocketWrapper to the

Processor

git-svn-id: https://svn.apache.org/repos/asf/tomcat/trunk@1709543 13f79535-47bb-0310-9956-ffa450edef68

java/org/apache/coyote/AbstractProcessorLight.java

@@ -16,11 +16,53 @@
  */
 package org.apache.coyote;
 
+import java.util.Iterator;
+import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
+
+import org.apache.tomcat.util.net.DispatchType;
+
 /**
  * This is a light-weight abstract processor implementation that is intended as
  * a basis for all Processor implementations from the light-weight upgrade
  * processors to the HTTP/AJP processors.
  */
 public abstract class AbstractProcessorLight implements Processor {
 
+    private Set<DispatchType> dispatches = new CopyOnWriteArraySet<>();
+
+
+    @Override
+    public void addDispatch(DispatchType dispatchType) {
+        synchronized (dispatches) {
+            dispatches.add(dispatchType);
+        }
+    }
+
+
+    @Override
+    public Iterator<DispatchType> getIteratorAndClearDispatches() {
+        // Note: Logic in AbstractProtocol depends on this method only returning
+        // a non-null value if the iterator is non-empty. i.e. it should never
+        // return an empty iterator.
+        Iterator<DispatchType> result;
+        synchronized (dispatches) {
+            // Synchronized as the generation of the iterator and the clearing
+            // of dispatches needs to be an atomic operation.
+            result = dispatches.iterator();
+            if (result.hasNext()) {
+                dispatches.clear();
+            } else {
+                result = null;
+            }
+        }
+        return result;
+    }
+
+
+    protected void clearDispatches() {
+        synchronized (dispatches) {
+            dispatches.clear();
+        }
+    }
 }

java/org/apache/coyote/AbstractProtocol.java

@@ -758,7 +758,7 @@ public SocketState process(SocketWrapperBase<S> wrapper,
                     if (dispatches == null || !dispatches.hasNext()) {
                         // Only returns non-null iterator if there are
                         // dispatches to process.
-                        dispatches = wrapper.getIteratorAndClearDispatches();
+                        dispatches = processor.getIteratorAndClearDispatches();
                     }
                 } while (state == SocketState.ASYNC_END ||
                         state == SocketState.UPGRADING ||

java/org/apache/coyote/Processor.java

@@ -18,11 +18,13 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.Iterator;
 import java.util.concurrent.Executor;
 
 import javax.servlet.http.HttpUpgradeHandler;
 
 import org.apache.tomcat.util.net.AbstractEndpoint.Handler.SocketState;
+import org.apache.tomcat.util.net.DispatchType;
 import org.apache.tomcat.util.net.SSLSupport;
 import org.apache.tomcat.util.net.SocketStatus;
 import org.apache.tomcat.util.net.SocketWrapperBase;
@@ -96,4 +98,8 @@ public interface Processor {
      * an existing multiplexed connection.
      */
     void pause();
+
+    void addDispatch(DispatchType dispatchType);
+
+    Iterator<DispatchType> getIteratorAndClearDispatches();
 }

java/org/apache/coyote/ajp/AjpProcessor.java

@@ -483,7 +483,7 @@ public final void action(ActionCode actionCode, Object param) {
             break;
         }
         case ASYNC_COMPLETE: {
-            socketWrapper.clearDispatches();
+            clearDispatches();
             if (asyncStateMachine.asyncComplete()) {
                 socketWrapper.processSocket(SocketStatus.OPEN_READ, true);
             }
@@ -573,15 +573,15 @@ public final void action(ActionCode actionCode, Object param) {
             break;
         }
         case DISPATCH_READ: {
-            socketWrapper.addDispatch(DispatchType.NON_BLOCKING_READ);
+            addDispatch(DispatchType.NON_BLOCKING_READ);
             break;
         }
         case DISPATCH_WRITE: {
-            socketWrapper.addDispatch(DispatchType.NON_BLOCKING_WRITE);
+            addDispatch(DispatchType.NON_BLOCKING_WRITE);
             break;
         }
         case DISPATCH_EXECUTE: {
-            socketWrapper.executeNonBlockingDispatches();
+            socketWrapper.executeNonBlockingDispatches(getIteratorAndClearDispatches());
             break;
         }
         case CLOSE_NOW: {

java/org/apache/coyote/http11/Http11Processor.java

@@ -731,7 +731,7 @@ public final void action(ActionCode actionCode, Object param) {
             break;
         }
         case ASYNC_COMPLETE: {
-            socketWrapper.clearDispatches();
+            clearDispatches();
             if (asyncStateMachine.asyncComplete()) {
                 socketWrapper.processSocket(SocketStatus.OPEN_READ, true);
             }
@@ -776,17 +776,17 @@ public final void action(ActionCode actionCode, Object param) {
             break;
         }
         case DISPATCH_READ: {
-            socketWrapper.addDispatch(DispatchType.NON_BLOCKING_READ);
+            addDispatch(DispatchType.NON_BLOCKING_READ);
             break;
         }
         case DISPATCH_WRITE: {
-            socketWrapper.addDispatch(DispatchType.NON_BLOCKING_WRITE);
+            addDispatch(DispatchType.NON_BLOCKING_WRITE);
             break;
         }
         case DISPATCH_EXECUTE: {
             SocketWrapperBase<?> wrapper = socketWrapper;
             if (wrapper != null) {
-                wrapper.executeNonBlockingDispatches();
+                wrapper.executeNonBlockingDispatches(getIteratorAndClearDispatches());
             }
             break;
         }

java/org/apache/coyote/http11/upgrade/UpgradeProcessorExternal.java

@@ -43,8 +43,8 @@ public class UpgradeProcessorExternal extends UpgradeProcessorBase {
     public UpgradeProcessorExternal(SocketWrapperBase<?> wrapper, ByteBuffer leftOverInput,
             HttpUpgradeHandler httpUpgradeHandler) {
         super(wrapper, leftOverInput, httpUpgradeHandler);
-        this.upgradeServletInputStream = new UpgradeServletInputStream(wrapper);
-        this.upgradeServletOutputStream = new UpgradeServletOutputStream(wrapper);
+        this.upgradeServletInputStream = new UpgradeServletInputStream(this, wrapper);
+        this.upgradeServletOutputStream = new UpgradeServletOutputStream(this, wrapper);
 
         wrapper.unRead(leftOverInput);
         /*

java/org/apache/coyote/http11/upgrade/UpgradeServletInputStream.java

@@ -22,6 +22,7 @@
 import javax.servlet.ServletInputStream;
 
 import org.apache.coyote.ContainerThreadMarker;
+import org.apache.coyote.Processor;
 import org.apache.juli.logging.Log;
 import org.apache.juli.logging.LogFactory;
 import org.apache.tomcat.util.ExceptionUtils;
@@ -35,6 +36,7 @@ public class UpgradeServletInputStream extends ServletInputStream {
     private static final StringManager sm =
             StringManager.getManager(UpgradeServletInputStream.class);
 
+    private final Processor processor;
     private final SocketWrapperBase<?> socketWrapper;
 
     private volatile boolean closed = false;
@@ -45,7 +47,8 @@ public class UpgradeServletInputStream extends ServletInputStream {
     private volatile ClassLoader applicationLoader = null;
 
 
-    public UpgradeServletInputStream(SocketWrapperBase<?> socketWrapper) {
+    public UpgradeServletInputStream(Processor processor, SocketWrapperBase<?> socketWrapper) {
+        this.processor = processor;
         this.socketWrapper = socketWrapper;
     }
 
@@ -101,7 +104,7 @@ public final void setReadListener(ReadListener listener) {
 
         // Container is responsible for first call to onDataAvailable().
         if (ContainerThreadMarker.isContainerThread()) {
-            socketWrapper.addDispatch(DispatchType.NON_BLOCKING_READ);
+            processor.addDispatch(DispatchType.NON_BLOCKING_READ);
         } else {
             socketWrapper.registerReadInterest();
         }

java/org/apache/coyote/http11/upgrade/UpgradeServletOutputStream.java

@@ -22,6 +22,7 @@
 import javax.servlet.WriteListener;
 
 import org.apache.coyote.ContainerThreadMarker;
+import org.apache.coyote.Processor;
 import org.apache.juli.logging.Log;
 import org.apache.juli.logging.LogFactory;
 import org.apache.tomcat.util.ExceptionUtils;
@@ -35,6 +36,7 @@ public class UpgradeServletOutputStream extends ServletOutputStream {
     private static final StringManager sm =
             StringManager.getManager(UpgradeServletOutputStream.class);
 
+    private final Processor processor;
     private final SocketWrapperBase<?> socketWrapper;
 
     // Used to ensure that isReady() and onWritePossible() have a consistent
@@ -61,7 +63,8 @@ public class UpgradeServletOutputStream extends ServletOutputStream {
     private volatile ClassLoader applicationLoader = null;
 
 
-    public UpgradeServletOutputStream(SocketWrapperBase<?> socketWrapper) {
+    public UpgradeServletOutputStream(Processor processor, SocketWrapperBase<?> socketWrapper) {
+        this.processor = processor;
         this.socketWrapper = socketWrapper;
     }
 
@@ -115,7 +118,7 @@ public final void setWriteListener(WriteListener listener) {
             registered = true;
             // Container is responsible for first call to onDataAvailable().
             if (ContainerThreadMarker.isContainerThread()) {
-                socketWrapper.addDispatch(DispatchType.NON_BLOCKING_WRITE);
+                processor.addDispatch(DispatchType.NON_BLOCKING_WRITE);
             } else {
                 socketWrapper.registerWriteInterest();
             }

java/org/apache/coyote/http2/StreamProcessor.java

@@ -19,8 +19,6 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Iterator;
-import java.util.Set;
-import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.servlet.http.HttpUpgradeHandler;
@@ -48,7 +46,6 @@ public class StreamProcessor extends AbstractProcessor implements Runnable {
     private static final StringManager sm = StringManager.getManager(StreamProcessor.class);
 
     private final Stream stream;
-    private Set<DispatchType> dispatches = new CopyOnWriteArraySet<>();
 
     private volatile SSLSupport sslSupport;
 
@@ -374,11 +371,11 @@ public void action(ActionCode actionCode, Object param) {
             break;
         }
         case DISPATCH_READ: {
-            dispatches.add(DispatchType.NON_BLOCKING_READ);
+            addDispatch(DispatchType.NON_BLOCKING_READ);
             break;
         }
         case DISPATCH_WRITE: {
-            dispatches.add(DispatchType.NON_BLOCKING_WRITE);
+            addDispatch(DispatchType.NON_BLOCKING_WRITE);
             break;
         }
         case DISPATCH_EXECUTE: {
@@ -482,35 +479,6 @@ protected SocketState dispatchEndRequest() {
     }
 
 
-    public void addDispatch(DispatchType dispatchType) {
-        synchronized (dispatches) {
-            dispatches.add(dispatchType);
-        }
-    }
-    public Iterator<DispatchType> getIteratorAndClearDispatches() {
-        // Note: Logic in AbstractProtocol depends on this method only returning
-        // a non-null value if the iterator is non-empty. i.e. it should never
-        // return an empty iterator.
-        Iterator<DispatchType> result;
-        synchronized (dispatches) {
-            // Synchronized as the generation of the iterator and the clearing
-            // of dispatches needs to be an atomic operation.
-            result = dispatches.iterator();
-            if (result.hasNext()) {
-                dispatches.clear();
-            } else {
-                result = null;
-            }
-        }
-        return result;
-    }
-    public void clearDispatches() {
-        synchronized (dispatches) {
-            dispatches.clear();
-        }
-    }
-
-
     @Override
     public HttpUpgradeHandler getHttpUpgradeHandler() {
         // Should never happen

java/org/apache/tomcat/util/net/AbstractEndpoint.java

@@ -811,7 +811,8 @@ public abstract void processSocket(SocketWrapperBase<S> socketWrapper,
             SocketStatus socketStatus, boolean dispatch);
 
 
-    public void executeNonBlockingDispatches(SocketWrapperBase<S> socketWrapper) {
+    public void executeNonBlockingDispatches(SocketWrapperBase<S> socketWrapper,
+            Iterator<DispatchType> dispatches) {
         /*
          * This method is called when non-blocking IO is initiated by defining
          * a read and/or write listener in a non-container thread. It is called
@@ -831,8 +832,6 @@ public void executeNonBlockingDispatches(SocketWrapperBase<S> socketWrapper) {
          * sure that the socket has been added to the waitingRequests queue.
          */
         synchronized (socketWrapper) {
-            Iterator<DispatchType> dispatches = socketWrapper.getIteratorAndClearDispatches();
-
             while (dispatches != null && dispatches.hasNext()) {
                 DispatchType dispatchType = dispatches.next();
                 processSocket(socketWrapper, dispatchType.getSocketStatus(), false);

java/org/apache/tomcat/util/net/SocketWrapperBase.java

@@ -20,8 +20,6 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.CompletionHandler;
 import java.util.Iterator;
-import java.util.Set;
-import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
@@ -98,8 +96,6 @@ public abstract class SocketWrapperBase<E> {
      */
     protected int bufferedWriteSize = 64*1024; //64k default write buffer
 
-    private Set<DispatchType> dispatches = new CopyOnWriteArraySet<>();
-
     public SocketWrapperBase(E socket, AbstractEndpoint<E> endpoint) {
         this.socket = socket;
         this.endpoint = endpoint;
@@ -298,34 +294,6 @@ public boolean canWrite() {
         return socketBufferHandler.isWriteBufferWritable() && bufferedWrites.size() == 0;
     }
 
-    public void addDispatch(DispatchType dispatchType) {
-        synchronized (dispatches) {
-            dispatches.add(dispatchType);
-        }
-    }
-    public Iterator<DispatchType> getIteratorAndClearDispatches() {
-        // Note: Logic in AbstractProtocol depends on this method only returning
-        // a non-null value if the iterator is non-empty. i.e. it should never
-        // return an empty iterator.
-        Iterator<DispatchType> result;
-        synchronized (dispatches) {
-            // Synchronized as the generation of the iterator and the clearing
-            // of dispatches needs to be an atomic operation.
-            result = dispatches.iterator();
-            if (result.hasNext()) {
-                dispatches.clear();
-            } else {
-                result = null;
-            }
-        }
-        return result;
-    }
-    public void clearDispatches() {
-        synchronized (dispatches) {
-            dispatches.clear();
-        }
-    }
-
 
     /**
      * Overridden for debug purposes. No guarantees are made about the format of
@@ -587,8 +555,8 @@ public void processSocket(SocketStatus socketStatus, boolean dispatch) {
     }
 
 
-    public void executeNonBlockingDispatches() {
-        endpoint.executeNonBlockingDispatches(this);
+    public void executeNonBlockingDispatches(Iterator<DispatchType> dispatches) {
+        endpoint.executeNonBlockingDispatches(this, dispatches);
     }