From: jfclere Date: Thu, 27 May 2010 12:05:50 +0000 (+0000) Subject: Fix APR AJP logic. X-Git-Url: https://git.internetallee.de/?a=commitdiff_plain;h=a0b3c24d420172e98298769394264f60c9707668;p=tomcat7.0 Fix APR AJP logic. git-svn-id: https://svn.apache.org/repos/asf/tomcat/trunk@948795 13f79535-47bb-0310-9956-ffa450edef68 --- diff --git a/java/org/apache/coyote/ajp/AjpAprProcessor.java b/java/org/apache/coyote/ajp/AjpAprProcessor.java index 7159bcd96..b79c7301c 100644 --- a/java/org/apache/coyote/ajp/AjpAprProcessor.java +++ b/java/org/apache/coyote/ajp/AjpAprProcessor.java @@ -22,6 +22,7 @@ import java.io.IOException; import java.io.InterruptedIOException; import java.net.InetAddress; import java.nio.ByteBuffer; +import java.util.concurrent.atomic.AtomicBoolean; import java.security.cert.CertificateFactory; import java.security.cert.X509Certificate; @@ -44,6 +45,8 @@ import org.apache.tomcat.util.http.HttpMessages; import org.apache.tomcat.util.http.MimeHeaders; import org.apache.tomcat.util.net.AbstractEndpoint; import org.apache.tomcat.util.net.AprEndpoint; +import org.apache.tomcat.util.net.SocketStatus; +import org.apache.tomcat.util.net.AbstractEndpoint.Handler.SocketState; import org.apache.tomcat.util.res.StringManager; @@ -183,6 +186,10 @@ public class AjpAprProcessor implements ActionHook { */ protected boolean error = false; + /** + * Async used + */ + protected boolean async = false; /** * Socket associated with the current connection. @@ -366,6 +373,7 @@ public class AjpAprProcessor implements ActionHook { // Error flag error = false; + async = false; boolean openSocket = true; boolean keptAlive = false; @@ -437,6 +445,10 @@ public class AjpAprProcessor implements ActionHook { } } + if (async && !error) { + break; + } + // Finish the response if not done yet if (!finished) { try { @@ -466,12 +478,53 @@ public class AjpAprProcessor implements ActionHook { } rp.setStage(org.apache.coyote.Constants.STAGE_ENDED); - recycle(); + if (!async || error) + recycle(); return openSocket; } + /* Copied from the AjpProcessor.java */ + public SocketState asyncDispatch(long socket, SocketStatus status) throws IOException { + + // Setting up the socket + this.socket = socket; + + RequestInfo rp = request.getRequestProcessor(); + try { + rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE); + error = !adapter.asyncDispatch(request, response, status); + } catch (InterruptedIOException e) { + error = true; + } catch (Throwable t) { + log.error(sm.getString("http11processor.request.process"), t); + // 500 - Internal Server Error + response.setStatus(500); + error = true; + } + + rp.setStage(org.apache.coyote.Constants.STAGE_ENDED); + + if (async) { + if (error) { + response.setStatus(500); + request.updateCounters(); + recycle(); + return SocketState.CLOSED; + } else { + return SocketState.LONG; + } + } else { + if (error) { + response.setStatus(500); + } + request.updateCounters(); + recycle(); + return SocketState.CLOSED; + } + + } // ----------------------------------------------------- ActionHook Methods @@ -524,6 +577,7 @@ public class AjpAprProcessor implements ActionHook { } else if (actionCode == ActionCode.ACTION_CLOSE) { // Close + async = false; // End the processing of the current request, and stop any further // transactions with the client @@ -604,6 +658,31 @@ public class AjpAprProcessor implements ActionHook { empty = false; replay = true; + } else if (actionCode == ActionCode.ACTION_ASYNC_START) { + async = true; + } else if (actionCode == ActionCode.ACTION_ASYNC_COMPLETE) { + AtomicBoolean dispatch = (AtomicBoolean)param; + RequestInfo rp = request.getRequestProcessor(); + if ( rp.getStage() != org.apache.coyote.Constants.STAGE_SERVICE ) { //async handling + dispatch.set(true); + endpoint.getHandler().asyncDispatch(this.socket, SocketStatus.STOP); + } else { + dispatch.set(false); + } + } else if (actionCode == ActionCode.ACTION_ASYNC_SETTIMEOUT) { + if (param==null) return; + if (socket==0) return; + long timeout = ((Long)param).longValue(); + Socket.timeoutSet(socket, timeout * 1000); + } else if (actionCode == ActionCode.ACTION_ASYNC_DISPATCH) { + RequestInfo rp = request.getRequestProcessor(); + AtomicBoolean dispatch = (AtomicBoolean)param; + if ( rp.getStage() != org.apache.coyote.Constants.STAGE_SERVICE ) {//async handling + endpoint.getPoller().add(this.socket); + dispatch.set(true); + } else { + dispatch.set(true); + } } diff --git a/java/org/apache/coyote/ajp/AjpAprProtocol.java b/java/org/apache/coyote/ajp/AjpAprProtocol.java index b2862795e..cc1410e89 100644 --- a/java/org/apache/coyote/ajp/AjpAprProtocol.java +++ b/java/org/apache/coyote/ajp/AjpAprProtocol.java @@ -21,6 +21,7 @@ import java.net.InetAddress; import java.net.URLEncoder; import java.util.Hashtable; import java.util.Iterator; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.Executor; import java.util.concurrent.atomic.AtomicInteger; @@ -325,6 +326,9 @@ public class AjpAprProtocol protected AtomicLong registerCount = new AtomicLong(0); protected RequestGroupInfo global = new RequestGroupInfo(); + protected ConcurrentHashMap connections = + new ConcurrentHashMap(); + protected ConcurrentLinkedQueue recycledProcessors = new ConcurrentLinkedQueue() { protected AtomicInteger size = new AtomicInteger(0); @@ -384,8 +388,10 @@ public class AjpAprProtocol processor.action(ActionCode.ACTION_START, null); if (processor.process(socket)) { + connections.put(Long.valueOf(socket), processor); return SocketState.OPEN; } else { + // recycledProcessors.offer(processor); return SocketState.CLOSED; } @@ -418,7 +424,45 @@ public class AjpAprProtocol // FIXME: Support for this could be added in AJP as well public SocketState asyncDispatch(long socket, SocketStatus status) { - return SocketState.CLOSED; + + AjpAprProcessor result = connections.get(Long.valueOf(socket)); + + SocketState state = SocketState.CLOSED; + if (result != null) { + // Call the appropriate event + try { + state = result.asyncDispatch(socket, status); + } catch (java.net.SocketException e) { + // SocketExceptions are normal + AjpAprProtocol.log.debug + (sm.getString + ("ajpprotocol.proto.socketexception.debug"), e); + } catch (java.io.IOException e) { + // IOExceptions are normal + AjpAprProtocol.log.debug + (sm.getString + ("ajpprotocol.proto.ioexception.debug"), e); + } + // Future developers: if you discover any other + // rare-but-nonfatal exceptions, catch them here, and log as + // above. + catch (Throwable e) { + // any other exception or error is odd. Here we log it + // with "ERROR" level, so it will show up even on + // less-than-verbose logs. + AjpAprProtocol.log.error + (sm.getString("ajpprotocol.proto.error"), e); + } finally { + if (state != SocketState.LONG) { + connections.remove(Long.valueOf(socket)); + recycledProcessors.offer(result); + if (state == SocketState.OPEN) { + proto.endpoint.getPoller().add(socket); + } + } + } + } + return state; } protected AjpAprProcessor createProcessor() {