tomcat-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From r...@apache.org
Subject svn commit: r407241 - in /tomcat/tc6.0.x/trunk/java/org/apache: catalina/ catalina/connector/ catalina/servlets/ coyote/ajp/ coyote/http11/ tomcat/util/net/
Date Wed, 17 May 2006 12:55:40 GMT
Author: remm
Date: Wed May 17 05:55:39 2006
New Revision: 407241

URL: http://svn.apache.org/viewcvs?rev=407241&view=rev
Log:
- Start work on comet support. Note: it doesn't work yet, I think (I didn't test), and most
of this
  is very preliminary. It is relatively straightforward, though.

Added:
    tomcat/tc6.0.x/trunk/java/org/apache/catalina/CometProcessor.java   (with props)
    tomcat/tc6.0.x/trunk/java/org/apache/catalina/servlets/CometServlet.java   (with props)
Modified:
    tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/CoyoteAdapter.java
    tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/Request.java
    tomcat/tc6.0.x/trunk/java/org/apache/coyote/ajp/AjpAprProtocol.java
    tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProcessor.java
    tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProtocol.java
    tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/InternalAprInputBuffer.java
    tomcat/tc6.0.x/trunk/java/org/apache/tomcat/util/net/AprEndpoint.java

Added: tomcat/tc6.0.x/trunk/java/org/apache/catalina/CometProcessor.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/catalina/CometProcessor.java?rev=407241&view=auto
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/catalina/CometProcessor.java (added)
+++ tomcat/tc6.0.x/trunk/java/org/apache/catalina/CometProcessor.java Wed May 17 05:55:39
2006
@@ -0,0 +1,21 @@
+package org.apache.catalina;
+
+import java.io.IOException;
+
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+public interface CometProcessor {
+
+    public void begin(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException;
+    public void end(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException;
+
+    public void error(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException;
+    public void read(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException;
+
+}

Propchange: tomcat/tc6.0.x/trunk/java/org/apache/catalina/CometProcessor.java
------------------------------------------------------------------------------
    svn:eol-style = native

Modified: tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/CoyoteAdapter.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/CoyoteAdapter.java?rev=407241&r1=407240&r2=407241&view=diff
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/CoyoteAdapter.java (original)
+++ tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/CoyoteAdapter.java Wed May 17
05:55:39 2006
@@ -19,6 +19,7 @@
 
 import java.io.IOException;
 
+import org.apache.catalina.CometProcessor;
 import org.apache.catalina.Context;
 import org.apache.catalina.Globals;
 import org.apache.catalina.Wrapper;
@@ -135,10 +136,35 @@
 
         }
 
+        // Comet processing
+        if (request.getWrapper() != null 
+                && request.getWrapper() instanceof CometProcessor) {
+            try {
+                if (request.getAttribute("org.apache.tomcat.comet.error") != null) {
+                    ((CometProcessor) request.getWrapper()).error(request.getRequest(), response.getResponse());
+                } else {
+                    ((CometProcessor) request.getWrapper()).read(request.getRequest(), response.getResponse());
+                }
+            } catch (IOException e) {
+                ;
+            } catch (Throwable t) {
+                log.error(sm.getString("coyoteAdapter.service"), t);
+            } finally {
+                // Recycle the wrapper request and response
+                if (request.getAttribute("org.apache.tomcat.comet") == null) {
+                    request.recycle();
+                    response.recycle();
+                }
+            }
+            return;
+        }
+        
         if (connector.getXpoweredBy()) {
             response.addHeader("X-Powered-By", "Servlet/2.5");
         }
 
+        boolean comet = false;
+        
         try {
 
             // Parse and set Catalina and configuration specific 
@@ -148,8 +174,16 @@
                 connector.getContainer().getPipeline().getFirst().invoke(request, response);
             }
 
-            response.finishResponse();
-            req.action( ActionCode.ACTION_POST_REQUEST , null);
+            if (request.getAttribute("org.apache.tomcat.comet.support") == Boolean.TRUE 
+                    && request.getWrapper() instanceof CometProcessor) {
+                request.setAttribute("org.apache.tomcat.comet", Boolean.TRUE);
+                comet = true;
+            }
+
+            if (!comet) {
+                response.finishResponse();
+                req.action( ActionCode.ACTION_POST_REQUEST , null);
+            }
 
         } catch (IOException e) {
             ;
@@ -157,8 +191,10 @@
             log.error(sm.getString("coyoteAdapter.service"), t);
         } finally {
             // Recycle the wrapper request and response
-            request.recycle();
-            response.recycle();
+            if (!comet) {
+                request.recycle();
+                response.recycle();
+            }
         }
 
     }

Modified: tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/Request.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/Request.java?rev=407241&r1=407240&r2=407241&view=diff
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/Request.java (original)
+++ tomcat/tc6.0.x/trunk/java/org/apache/catalina/connector/Request.java Wed May 17 05:55:39
2006
@@ -1294,6 +1294,12 @@
         if (readOnlyAttributes.containsKey(name)) {
             return;
         }
+
+        // Pass special attributes to the native layer
+        if (name.startsWith("org.apache.tomcat.")) {
+            coyoteRequest.getAttributes().remove(name);
+        }
+
         found = attributes.containsKey(name);
         if (found) {
             value = attributes.get(name);
@@ -1301,7 +1307,7 @@
         } else {
             return;
         }
-
+        
         // Notify interested application event listeners
         Object listeners[] = context.getApplicationEventListeners();
         if ((listeners == null) || (listeners.length == 0))

Added: tomcat/tc6.0.x/trunk/java/org/apache/catalina/servlets/CometServlet.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/catalina/servlets/CometServlet.java?rev=407241&view=auto
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/catalina/servlets/CometServlet.java (added)
+++ tomcat/tc6.0.x/trunk/java/org/apache/catalina/servlets/CometServlet.java Wed May 17 05:55:39
2006
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2006 The Apache Software Foundation.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.catalina.servlets;
+
+
+import java.io.IOException;
+
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.catalina.CometProcessor;
+
+
+/**
+ * Helper class to implement Comet functionality.
+ */
+public abstract class CometServlet
+    extends HttpServlet implements CometProcessor {
+
+    public void begin(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException {
+        
+    }
+    
+    public void end(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException {
+        request.removeAttribute("org.apache.tomcat.comet");
+    }
+    
+    public void error(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException {
+        end(request, response);
+    }
+    
+    public abstract void read(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException;
+
+    protected void service(HttpServletRequest request, HttpServletResponse response)
+        throws IOException, ServletException {
+        
+        if (request.getAttribute("org.apache.tomcat.comet.support") == Boolean.TRUE) {
+            begin(request, response);
+        } else {
+            // FIXME: Implement without comet support
+            begin(request, response);
+            
+            // Loop reading data
+            
+            end(request, response);
+        }
+        
+    }
+
+}

Propchange: tomcat/tc6.0.x/trunk/java/org/apache/catalina/servlets/CometServlet.java
------------------------------------------------------------------------------
    svn:eol-style = native

Modified: tomcat/tc6.0.x/trunk/java/org/apache/coyote/ajp/AjpAprProtocol.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/coyote/ajp/AjpAprProtocol.java?rev=407241&r1=407240&r2=407241&view=diff
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/coyote/ajp/AjpAprProtocol.java (original)
+++ tomcat/tc6.0.x/trunk/java/org/apache/coyote/ajp/AjpAprProtocol.java Wed May 17 05:55:39
2006
@@ -35,6 +35,7 @@
 import org.apache.tomcat.util.modeler.Registry;
 import org.apache.tomcat.util.net.AprEndpoint;
 import org.apache.tomcat.util.net.AprEndpoint.Handler;
+import org.apache.tomcat.util.net.AprEndpoint.Handler.SocketState;
 import org.apache.tomcat.util.res.StringManager;
 
 
@@ -429,7 +430,12 @@
             this.proto = proto;
         }
 
-        public boolean process(long socket) {
+        // FIXME: Support for this could be added in AJP as well
+        public SocketState event(long socket, boolean error) {
+            return SocketState.CLOSED;
+        }
+        
+        public SocketState process(long socket) {
             AjpAprProcessor processor = null;
             try {
                 processor = (AjpAprProcessor) localProcessor.get();
@@ -460,7 +466,11 @@
                     ((ActionHook) processor).action(ActionCode.ACTION_START, null);
                 }
 
-                return processor.process(socket);
+                if (processor.process(socket)) {
+                    return SocketState.OPEN;
+                } else {
+                    return SocketState.CLOSED;
+                }
 
             } catch(java.net.SocketException e) {
                 // SocketExceptions are normal
@@ -487,7 +497,7 @@
                     ((ActionHook) processor).action(ActionCode.ACTION_STOP, null);
                 }
             }
-            return false;
+            return SocketState.CLOSED;
         }
     }
 

Modified: tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProcessor.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProcessor.java?rev=407241&r1=407240&r2=407241&view=diff
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProcessor.java (original)
+++ tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProcessor.java Wed May 17
05:55:39 2006
@@ -52,6 +52,7 @@
 import org.apache.tomcat.util.http.FastHttpDateFormat;
 import org.apache.tomcat.util.http.MimeHeaders;
 import org.apache.tomcat.util.net.AprEndpoint;
+import org.apache.tomcat.util.net.AprEndpoint.Handler.SocketState;
 import org.apache.tomcat.util.res.StringManager;
 
 
@@ -147,12 +148,6 @@
 
 
     /**
-     * State flag.
-     */
-    protected boolean started = false;
-
-
-    /**
      * Error flag.
      */
     protected boolean error = false;
@@ -183,6 +178,12 @@
 
 
     /**
+     * Comet used.
+     */
+    protected boolean comet = false;
+
+
+    /**
      * Content delimitator for the request (if false, the connection will
      * be closed at the end of the request).
      */
@@ -735,7 +736,53 @@
      *
      * @throws IOException error during an I/O operation
      */
-    public boolean process(long socket)
+    public SocketState event(boolean error)
+        throws IOException {
+        
+        RequestInfo rp = request.getRequestProcessor();
+        
+        try {
+            rp.setStage(org.apache.coyote.Constants.STAGE_SERVICE);
+            if (error) {
+                request.setAttribute("org.apache.tomcat.comet.error", Boolean.TRUE);
+            }
+            // FIXME: It is also possible to add a new "event" method in the adapter
+            // or something similar
+            adapter.service(request, response);
+            if (request.getAttribute("org.apache.tomcat.comet") == null) {
+                comet = false;
+                endpoint.getCometPoller().remove(socket);
+            }
+        } 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 (error) {
+            recycle();
+            return SocketState.CLOSED;
+        } else if (!comet) {
+            recycle();
+            endpoint.getPoller().add(socket);
+            return SocketState.OPEN;
+        } else {
+            return SocketState.LONG;
+        }
+    }
+    
+    /**
+     * Process pipelined HTTP requests using the specified input and output
+     * streams.
+     *
+     * @throws IOException error during an I/O operation
+     */
+    public SocketState process(long socket)
         throws IOException {
         RequestInfo rp = request.getRequestProcessor();
         rp.setStage(org.apache.coyote.Constants.STAGE_PARSE);
@@ -768,7 +815,7 @@
         boolean keptAlive = false;
         boolean openSocket = false;
 
-        while (started && !error && keepAlive) {
+        while (!error && keepAlive) {
 
             // Parsing the request header
             try {
@@ -833,7 +880,10 @@
                         error = response.getErrorException() != null ||
                                 statusDropsConnection(response.getStatus());
                     }
-
+                    // Comet support
+                    if (request.getAttribute("org.apache.tomcat.comet") != null) {
+                        comet = true;
+                    }
                 } catch (InterruptedIOException e) {
                     error = true;
                 } catch (Throwable t) {
@@ -845,25 +895,8 @@
             }
 
             // Finish the handling of the request
-            try {
-                rp.setStage(org.apache.coyote.Constants.STAGE_ENDINPUT);
-                inputBuffer.endRequest();
-            } catch (IOException e) {
-                error = true;
-            } catch (Throwable t) {
-                log.error(sm.getString("http11processor.request.finish"), t);
-                // 500 - Internal Server Error
-                response.setStatus(500);
-                error = true;
-            }
-            try {
-                rp.setStage(org.apache.coyote.Constants.STAGE_ENDOUTPUT);
-                outputBuffer.endRequest();
-            } catch (IOException e) {
-                error = true;
-            } catch (Throwable t) {
-                log.error(sm.getString("http11processor.response.finish"), t);
-                error = true;
+            if (!comet) {
+                endRequest();
             }
 
             // If there was an error, make sure the request is counted as
@@ -873,17 +906,8 @@
             }
             request.updateCounters();
 
-            rp.setStage(org.apache.coyote.Constants.STAGE_KEEPALIVE);
-
-            // Don't reset the param - we'll see it as ended. Next request
-            // will reset it
-            // thrA.setParam(null);
-            // Next request
-            inputBuffer.nextRequest();
-            outputBuffer.nextRequest();
-
             // Do sendfile as needed: add socket to sendfile and end
-            if (sendfileData != null) {
+            if (sendfileData != null && !error) {
                 sendfileData.socket = socket;
                 sendfileData.keepAlive = keepAlive;
                 if (!endpoint.getSendfile().add(sendfileData)) {
@@ -892,19 +916,63 @@
                 }
             }
             
+            rp.setStage(org.apache.coyote.Constants.STAGE_KEEPALIVE);
+
         }
 
         rp.setStage(org.apache.coyote.Constants.STAGE_ENDED);
 
-        // Recycle
+        if (comet) {
+            if (error) {
+                recycle();
+                return SocketState.CLOSED;
+            } else {
+                endpoint.getCometPoller().add(socket);
+                return SocketState.LONG;
+            }
+        } else {
+            recycle();
+            return (openSocket) ? SocketState.OPEN : SocketState.CLOSED;
+        }
+        
+    }
+
+    
+    public void endRequest() {
+        
+        // Finish the handling of the request
+        try {
+            inputBuffer.endRequest();
+        } catch (IOException e) {
+            error = true;
+        } catch (Throwable t) {
+            log.error(sm.getString("http11processor.request.finish"), t);
+            // 500 - Internal Server Error
+            response.setStatus(500);
+            error = true;
+        }
+        try {
+            outputBuffer.endRequest();
+        } catch (IOException e) {
+            error = true;
+        } catch (Throwable t) {
+            log.error(sm.getString("http11processor.response.finish"), t);
+            error = true;
+        }
+
+        // Next request
+        inputBuffer.nextRequest();
+        outputBuffer.nextRequest();
+        
+    }
+    
+    
+    public void recycle() {
         inputBuffer.recycle();
         outputBuffer.recycle();
         this.socket = 0;
-
-        return openSocket;
-        
     }
-
+    
 
     // ----------------------------------------------------- ActionHook Methods
 
@@ -966,6 +1034,7 @@
             // End the processing of the current request, and stop any further
             // transactions with the client
 
+            comet = false;
             try {
                 outputBuffer.endRequest();
             } catch (IOException e) {
@@ -985,14 +1054,6 @@
 
             // Do nothing
 
-        } else if (actionCode == ActionCode.ACTION_START) {
-
-            started = true;
-
-        } else if (actionCode == ActionCode.ACTION_STOP) {
-
-            started = false;
-
         } else if (actionCode == ActionCode.ACTION_REQ_HOST_ADDR_ATTRIBUTE) {
 
             // Get remote host address
@@ -1368,6 +1429,8 @@
         if (endpoint.getUseSendfile()) {
             request.setAttribute("org.apache.tomcat.sendfile.support", Boolean.TRUE);
         }
+        // Advertise comet support through a request attribute
+        request.setAttribute("org.apache.tomcat.comet.support", Boolean.TRUE);
         
     }
 

Modified: tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProtocol.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProtocol.java?rev=407241&r1=407240&r2=407241&view=diff
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProtocol.java (original)
+++ tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/Http11AprProtocol.java Wed May 17 05:55:39
2006
@@ -20,6 +20,7 @@
 import java.net.URLEncoder;
 import java.util.Hashtable;
 import java.util.Iterator;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 
 import javax.management.MBeanRegistration;
@@ -598,20 +599,73 @@
     // --------------------  Connection handler --------------------
 
     static class Http11ConnectionHandler implements Handler {
-        Http11AprProtocol proto;
-        static int count=0;
-        RequestGroupInfo global=new RequestGroupInfo();
-        ThreadLocal localProcessor = new ThreadLocal();
+        
+        protected Http11AprProtocol proto;
+        protected static int count = 0;
+        protected RequestGroupInfo global = new RequestGroupInfo();
+        
+        protected ThreadLocal<Http11AprProcessor> localProcessor = 
+            new ThreadLocal<Http11AprProcessor>();
+        protected ConcurrentHashMap<Long, Http11AprProcessor> connections =
+            new ConcurrentHashMap<Long, Http11AprProcessor>();
+        protected java.util.Stack<Http11AprProcessor> recycledProcessors = 
+            new java.util.Stack<Http11AprProcessor>();
 
-        Http11ConnectionHandler( Http11AprProtocol proto ) {
-            this.proto=proto;
+        Http11ConnectionHandler(Http11AprProtocol proto) {
+            this.proto = proto;
         }
 
-        public boolean process(long socket) {
+        public SocketState event(long socket, boolean error) {
+            Http11AprProcessor result = connections.get(socket);
+            SocketState state = SocketState.CLOSED; 
+            if (result != null) {
+                boolean recycle = error;
+                // Call the appropriate event
+                try {
+                    state = result.event(error);
+                } catch (java.net.SocketException e) {
+                    // SocketExceptions are normal
+                    Http11AprProtocol.log.debug
+                        (sm.getString
+                            ("http11protocol.proto.socketexception.debug"), e);
+                } catch (java.io.IOException e) {
+                    // IOExceptions are normal
+                    Http11AprProtocol.log.debug
+                        (sm.getString
+                            ("http11protocol.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.
+                    Http11AprProtocol.log.error
+                        (sm.getString("http11protocol.proto.error"), e);
+                } finally {
+                    if (state != SocketState.LONG) {
+                        connections.remove(socket);
+                        recycledProcessors.push(result);
+                    }
+                }
+            }
+            return state;
+        }
+        
+        public SocketState process(long socket) {
             Http11AprProcessor processor = null;
             try {
                 processor = (Http11AprProcessor) localProcessor.get();
                 if (processor == null) {
+                    synchronized (recycledProcessors) {
+                        if (!recycledProcessors.isEmpty()) {
+                            processor = recycledProcessors.pop();
+                            localProcessor.set(processor);
+                        }
+                    }
+                }
+                if (processor == null) {
                     processor =
                         new Http11AprProcessor(proto.maxHttpHeaderSize, proto.ep);
                     processor.setAdapter(proto.adapter);
@@ -647,7 +701,15 @@
                     ((ActionHook) processor).action(ActionCode.ACTION_START, null);
                 }
 
-                return processor.process(socket);
+                SocketState state = processor.process(socket);
+                if (state == SocketState.LONG) {
+                    // Associate the connection with the processor. The next request 
+                    // processed by this thread will use either a new or a recycled
+                    // processor.
+                    connections.put(socket, processor);
+                    localProcessor.set(null);
+                }
+                return state;
 
             } catch(java.net.SocketException e) {
                 // SocketExceptions are normal
@@ -669,15 +731,8 @@
                 // less-than-verbose logs.
                 Http11AprProtocol.log.error
                     (sm.getString("http11protocol.proto.error"), e);
-            } finally {
-                //       if(proto.adapter != null) proto.adapter.recycle();
-                //                processor.recycle();
-
-                if (processor instanceof ActionHook) {
-                    ((ActionHook) processor).action(ActionCode.ACTION_STOP, null);
-                }
             }
-            return false;
+            return SocketState.CLOSED;
         }
     }
 

Modified: tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/InternalAprInputBuffer.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/InternalAprInputBuffer.java?rev=407241&r1=407240&r2=407241&view=diff
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/InternalAprInputBuffer.java (original)
+++ tomcat/tc6.0.x/trunk/java/org/apache/coyote/http11/InternalAprInputBuffer.java Wed May
17 05:55:39 2006
@@ -329,8 +329,7 @@
      * consumed. This method only resets all the pointers so that we are ready
      * to parse the next HTTP request.
      */
-    public void nextRequest()
-        throws IOException {
+    public void nextRequest() {
 
         // Recycle Request object
         request.recycle();

Modified: tomcat/tc6.0.x/trunk/java/org/apache/tomcat/util/net/AprEndpoint.java
URL: http://svn.apache.org/viewcvs/tomcat/tc6.0.x/trunk/java/org/apache/tomcat/util/net/AprEndpoint.java?rev=407241&r1=407240&r2=407241&view=diff
==============================================================================
--- tomcat/tc6.0.x/trunk/java/org/apache/tomcat/util/net/AprEndpoint.java (original)
+++ tomcat/tc6.0.x/trunk/java/org/apache/tomcat/util/net/AprEndpoint.java Wed May 17 05:55:39
2006
@@ -300,6 +300,14 @@
 
 
     /**
+     * Allow comet request handling.
+     */
+    protected boolean useComet = true;
+    public void setUseComet(boolean useComet) { this.useComet = useComet; }
+    public boolean getUseComet() { return useComet; }
+
+
+    /**
      * Acceptor thread count.
      */
     protected int acceptorThreadCount = 0;
@@ -335,6 +343,17 @@
 
 
     /**
+     * The socket poller used for Comet support.
+     */
+    protected Poller[] cometPollers = null;
+    protected int cometPollerRoundRobin = 0;
+    public Poller getCometPoller() {
+        cometPollerRoundRobin = (cometPollerRoundRobin + 1) % cometPollers.length;
+        return cometPollers[cometPollerRoundRobin];
+    }
+
+
+    /**
      * The static file sender.
      */
     protected Sendfile[] sendfiles = null;
@@ -561,11 +580,8 @@
             addressStr = address.getHostAddress();
         }
         int family = Socket.APR_INET;
-        if (Library.APR_HAVE_IPV6) {
-            if (addressStr == null)
-                family = Socket.APR_UNSPEC;
-            else if (addressStr.indexOf(':') >= 0)
-                family = Socket.APR_UNSPEC;
+        if (Library.APR_HAVE_IPV6 && (addressStr == null || addressStr.indexOf(':')
>= 0)) {
+            family = Socket.APR_UNSPEC;
         }
         long inetAddress = Address.info(addressStr, family,
                 port, 0, rootPool);
@@ -712,7 +728,7 @@
             // Start poller threads
             pollers = new Poller[pollerThreadCount];
             for (int i = 0; i < pollerThreadCount; i++) {
-                pollers[i] = new Poller();
+                pollers[i] = new Poller(false);
                 pollers[i].init();
                 Thread pollerThread = new Thread(pollers[i], getName() + "-Poller-" + i);
                 pollerThread.setPriority(threadPriority);
@@ -720,6 +736,17 @@
                 pollerThread.start();
             }
 
+            // Start comet poller threads
+            cometPollers = new Poller[pollerThreadCount];
+            for (int i = 0; i < pollerThreadCount; i++) {
+                cometPollers[i] = new Poller(true);
+                cometPollers[i].init();
+                Thread pollerThread = new Thread(cometPollers[i], getName() + "-CometPoller-"
+ i);
+                pollerThread.setPriority(threadPriority);
+                pollerThread.setDaemon(true);
+                pollerThread.start();
+            }
+
             // Start sendfile threads
             if (useSendfile) {
                 sendfiles = new Sendfile[sendfileThreadCount];
@@ -998,6 +1025,26 @@
     }
     
 
+    /**
+     * Process given socket for an event.
+     */
+    protected boolean processSocket(long socket, boolean error) {
+        try {
+            if (executor == null) {
+                getWorkerThread().assign(socket, error);
+            } else {
+                executor.execute(new SocketEventProcessor(socket, error));
+            }
+        } catch (Throwable t) {
+            // This means we got an OOM or similar creating a thread, or that
+            // the pool and its queue are full
+            log.error(sm.getString("endpoint.process.fail"), t);
+            return false;
+        }
+        return true;
+    }
+    
+
     // --------------------------------------------------- Acceptor Inner Class
 
 
@@ -1060,10 +1107,18 @@
 
         protected long[] addS;
         protected int addCount = 0;
+        protected long[] removeS;
+        protected int removeCount = 0;
         
+        protected boolean comet = true;
+
         protected int keepAliveCount = 0;
         public int getKeepAliveCount() { return keepAliveCount; }
 
+        public Poller(boolean comet) {
+            this.comet = comet;
+        }
+        
         /**
          * Create the poller. With some versions of APR, the maximum poller size will
          * be 62 (reocmpiling APR is necessary to remove this limitation).
@@ -1071,19 +1126,29 @@
         protected void init() {
             pool = Pool.create(serverSockPool);
             int size = pollerSize / pollerThreadCount;
-            serverPollset = allocatePoller(size, pool, soTimeout);
+            int timeout = soTimeout;
+            if (comet) {
+                // FIXME: Find an appropriate timeout value, for now, "longer than usual"
+                // semms appropriate
+                timeout = soTimeout * 20;
+            }
+            serverPollset = allocatePoller(size, pool, timeout);
             if (serverPollset == 0 && size > 1024) {
                 size = 1024;
-                serverPollset = allocatePoller(size, pool, soTimeout);
+                serverPollset = allocatePoller(size, pool, timeout);
             }
             if (serverPollset == 0) {
                 size = 62;
-                serverPollset = allocatePoller(size, pool, soTimeout);
+                serverPollset = allocatePoller(size, pool, timeout);
             }
             desc = new long[size * 2];
             keepAliveCount = 0;
             addS = new long[size];
             addCount = 0;
+            if (comet) {
+                removeS = new long[size];
+            }
+            removeCount = 0;
         }
 
         /**
@@ -1092,18 +1157,32 @@
         protected void destroy() {
             // Close all sockets in the add queue
             for (int i = 0; i < addCount; i++) {
+                if (comet) {
+                    processSocket(addS[i], true);
+                }
                 Socket.destroy(addS[i]);
             }
+            // Close all sockets in the remove queue
+            for (int i = 0; i < removeCount; i++) {
+                if (comet) {
+                    processSocket(removeS[i], true);
+                }
+                Socket.destroy(removeS[i]);
+            }
             // Close all sockets still in the poller
             int rv = Poll.pollset(serverPollset, desc);
             if (rv > 0) {
                 for (int n = 0; n < rv; n++) {
+                    if (comet) {
+                        processSocket(desc[n*2+1], true);
+                    }
                     Socket.destroy(desc[n*2+1]);
                 }
             }
             Pool.destroy(pool);
             keepAliveCount = 0;
             addCount = 0;
+            removeCount = 0;
         }
 
         /**
@@ -1120,6 +1199,9 @@
                 // at most for pollTime before being polled
                 if (addCount >= addS.length) {
                     // Can't do anything: close the socket right away
+                    if (comet) {
+                        processSocket(socket, true);
+                    }
                     Socket.destroy(socket);
                     return;
                 }
@@ -1130,6 +1212,30 @@
         }
 
         /**
+         * Remove specified socket and associated pool from the poller. The socket will
+         * be added to a temporary array, and polled first after a maximum amount
+         * of time equal to pollTime (in most cases, latency will be much lower,
+         * however). Note that this is automatic, except if the poller is used for
+         * comet.
+         *
+         * @param socket to remove from the poller
+         */
+        public void remove(long socket) {
+            synchronized (this) {
+                // Add socket to the list. Newly added sockets will wait
+                // at most for pollTime before being polled
+                if (removeCount >= removeS.length) {
+                    // Normally, it cannot happen ...
+                    Socket.destroy(socket);
+                    return;
+                }
+                removeS[removeCount] = socket;
+                removeCount++;
+                this.notify();
+            }
+        }
+
+        /**
          * The background thread that listens for incoming TCP/IP connections and
          * hands them off to an appropriate processor.
          */
@@ -1171,23 +1277,41 @@
                                     keepAliveCount++;
                                 } else {
                                     // Can't do anything: close the socket right away
+                                    if (comet) {
+                                        processSocket(addS[i], true);
+                                    }
                                     Socket.destroy(addS[i]);
                                 }
                             }
                             addCount = 0;
                         }
                     }
+                    // Remove sockets which are waiting to the poller
+                    if (removeCount > 0) {
+                        synchronized (this) {
+                            for (int i = 0; i < removeCount; i++) {
+                                int rv = Poll.remove(serverPollset, removeS[i]);
+                            }
+                            removeCount = 0;
+                        }
+                    }
+
                     maintainTime += pollTime;
                     // Pool for the specified interval
-                    int rv = Poll.poll(serverPollset, pollTime, desc, true);
+                    int rv = Poll.poll(serverPollset, pollTime, desc, !comet);
                     if (rv > 0) {
                         keepAliveCount -= rv;
                         for (int n = 0; n < rv; n++) {
                             // Check for failed sockets and hand this socket off to a worker
                             if (((desc[n*2] & Poll.APR_POLLHUP) == Poll.APR_POLLHUP)
                                     || ((desc[n*2] & Poll.APR_POLLERR) == Poll.APR_POLLERR)
+                                    || (comet && (!processSocket(desc[n*2+1], false)))

                                     || (!processSocket(desc[n*2+1]))) {
                                 // Close socket and clear pool
+                                if (comet) {
+                                    processSocket(desc[n*2+1], true);
+                                    Poll.remove(serverPollset, desc[n*2+1]);
+                                }
                                 Socket.destroy(desc[n*2+1]);
                                 continue;
                             }
@@ -1215,6 +1339,11 @@
                             keepAliveCount -= rv;
                             for (int n = 0; n < rv; n++) {
                                 // Close socket and clear pool
+                                if (comet) {
+                                    // FIXME: should really close in case of timeout ?
+                                    // FIXME: maybe comet should use an extended timeout
+                                    processSocket(desc[n], true);
+                                }
                                 Socket.destroy(desc[n]);
                             }
                         }
@@ -1242,6 +1371,8 @@
         protected Thread thread = null;
         protected boolean available = false;
         protected long socket = 0;
+        protected boolean event = false;
+        protected boolean error = false;
 
 
         /**
@@ -1265,6 +1396,28 @@
 
             // Store the newly available Socket and notify our thread
             this.socket = socket;
+            event = false;
+            error = false;
+            available = true;
+            notifyAll();
+
+        }
+
+
+        protected synchronized void assign(long socket, boolean error) {
+
+            // Wait for the Processor to get the previous Socket
+            while (available) {
+                try {
+                    wait();
+                } catch (InterruptedException e) {
+                }
+            }
+
+            // Store the newly available Socket and notify our thread
+            this.socket = socket;
+            event = true;
+            this.error = error;
             available = true;
             notifyAll();
 
@@ -1310,7 +1463,11 @@
                     continue;
 
                 // Process the request from this socket
-                if (!handler.process(socket)) {
+                if ((event) && (handler.event(socket, error) == Handler.SocketState.CLOSED))
{
+                    // Close socket and pool
+                    Socket.destroy(socket);
+                    socket = 0;
+                } else if (handler.process(socket) == Handler.SocketState.CLOSED) {
                     // Close socket and pool
                     Socket.destroy(socket);
                     socket = 0;
@@ -1622,7 +1779,11 @@
      * thread local fields.
      */
     public interface Handler {
-        public boolean process(long socket);
+        public enum SocketState {
+            OPEN, CLOSED, LONG
+        }
+        public SocketState process(long socket);
+        public SocketState event(long socket, boolean error);
     }
 
 
@@ -1700,7 +1861,38 @@
         public void run() {
 
             // Process the request from this socket
-            if (!handler.process(socket)) {
+            if (handler.process(socket) == Handler.SocketState.CLOSED) {
+                // Close socket and pool
+                Socket.destroy(socket);
+                socket = 0;
+            }
+
+        }
+        
+    }
+    
+    
+    // --------------------------------------- SocketEventProcessor Inner Class
+
+
+    /**
+     * This class is the equivalent of the Worker, but will simply use in an
+     * external Executor thread pool.
+     */
+    protected class SocketEventProcessor implements Runnable {
+        
+        protected long socket = 0;
+        protected boolean error = false; 
+        
+        public SocketEventProcessor(long socket, boolean error) {
+            this.socket = socket;
+            this.error = error;
+        }
+
+        public void run() {
+
+            // Process the request from this socket
+            if (handler.event(socket, error) == Handler.SocketState.CLOSED) {
                 // Close socket and pool
                 Socket.destroy(socket);
                 socket = 0;



---------------------------------------------------------------------
To unsubscribe, e-mail: dev-unsubscribe@tomcat.apache.org
For additional commands, e-mail: dev-help@tomcat.apache.org


Mime
View raw message