Implement maxConnections for NIO connector

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

diff --git a/java/org/apache/tomcat/util/net/AbstractEndpoint.java b/java/org/apache/tomcat/util/net/AbstractEndpoint.java
index 6a4edf2..c3dcda0 100644 (file)
--- a/java/org/apache/tomcat/util/net/AbstractEndpoint.java
+++ b/java/org/apache/tomcat/util/net/AbstractEndpoint.java
@@ -32,6 +32,7 @@ import org.apache.juli.logging.Log;
 import org.apache.tomcat.util.IntrospectionUtils;
 import org.apache.tomcat.util.net.jsse.JSSESocketFactory;
 import org.apache.tomcat.util.res.StringManager;
+import org.apache.tomcat.util.threads.CounterLatch;
 import org.apache.tomcat.util.threads.ResizableExecutor;
 import org.apache.tomcat.util.threads.TaskQueue;
 import org.apache.tomcat.util.threads.TaskThreadFactory;
@@ -123,6 +124,11 @@ public abstract class AbstractEndpoint {
     protected volatile boolean internalExecutor = false;
 
     /**
+     * counter for nr of connections handled by an endpoint
+     */
+    protected volatile CounterLatch connectionCounterLatch = null;
+
+    /**
      * Socket properties
      */
     protected SocketProperties socketProperties = new SocketProperties();

diff --git a/java/org/apache/tomcat/util/net/JIoEndpoint.java b/java/org/apache/tomcat/util/net/JIoEndpoint.java
index ecfa472..1cacd90 100644 (file)
--- a/java/org/apache/tomcat/util/net/JIoEndpoint.java
+++ b/java/org/apache/tomcat/util/net/JIoEndpoint.java
@@ -109,7 +109,7 @@ public class JIoEndpoint extends AbstractEndpoint {
         return false;
     }
     
-    protected CounterLatch connectionCounterLatch = null;
+    
 
 
     // ------------------------------------------------ Handler Inner Interface

diff --git a/java/org/apache/tomcat/util/net/NioEndpoint.java b/java/org/apache/tomcat/util/net/NioEndpoint.java
index c48101e..ac56e62 100644 (file)
--- a/java/org/apache/tomcat/util/net/NioEndpoint.java
+++ b/java/org/apache/tomcat/util/net/NioEndpoint.java
@@ -57,6 +57,7 @@ import org.apache.tomcat.util.net.AbstractEndpoint.Handler.SocketState;
 import org.apache.tomcat.util.net.SecureNioChannel.ApplicationBufferHandler;
 import org.apache.tomcat.util.net.jsse.JSSESocketFactory;
 import org.apache.tomcat.util.net.jsse.NioX509KeyManager;
+import org.apache.tomcat.util.threads.CounterLatch;
 
 /**
  * NIO tailored thread pool, providing the following services:
@@ -567,6 +568,7 @@ public class NioEndpoint extends AbstractEndpoint {
             running = true;
             paused = false;
             
+            connectionCounterLatch = new CounterLatch(0, getMaxConnections()); 
             // Create worker collection
             if ( getExecutor() == null ) {
                 createExecutor();
@@ -598,6 +600,8 @@ public class NioEndpoint extends AbstractEndpoint {
      */
     @Override
     public void stopInternal() {
+        connectionCounterLatch.releaseAll();
+        connectionCounterLatch = null;
         if (!paused) {
             pause();
         }
@@ -808,6 +812,8 @@ public class NioEndpoint extends AbstractEndpoint {
                     break;
                 }
                 try {
+                    //if we have reached max connections, wait
+                    connectionCounterLatch.await();
                     // Accept the next incoming connection from the server socket
                     SocketChannel socket = serverSock.accept();
                     // Hand this socket off to an appropriate processor
@@ -824,7 +830,9 @@ public class NioEndpoint extends AbstractEndpoint {
                                 if (log.isDebugEnabled())
                                     log.debug("", ix);
                             }
-                        } 
+                        } else {
+                            connectionCounterLatch.countUp();
+                        }
                     }
                 } catch (SocketTimeoutException sx) {
                     //normal condition
@@ -1056,7 +1064,12 @@ public class NioEndpoint extends AbstractEndpoint {
                 if (key.channel().isOpen()) try {key.channel().close();}catch (Exception ignore){}
                 try {if (ka!=null) ka.getSocket().close(true);}catch (Exception ignore){}
                 try {if (ka!=null && ka.getSendfileData()!=null && ka.getSendfileData().fchannel!=null && ka.getSendfileData().fchannel.isOpen()) ka.getSendfileData().fchannel.close();}catch (Exception ignore){}
-                if (ka!=null) ka.reset();
+                if (ka!=null) {
+                    ka.reset();
+                    if (connectionCounterLatch.countDown()<0) {
+                        log.warn("Incorrect connection count, multiple cancel called on the same key?" );
+                    }
+                }
             } catch (Throwable e) {
                 ExceptionUtils.handleThrowable(e);
                 if ( log.isDebugEnabled() ) log.error("",e);

diff --git a/test/org/apache/catalina/connector/TestMaxConnections.java b/test/org/apache/catalina/connector/TestMaxConnections.java
index 7bd2491..e0b7412 100644 (file)
--- a/test/org/apache/catalina/connector/TestMaxConnections.java
+++ b/test/org/apache/catalina/connector/TestMaxConnections.java
@@ -27,16 +27,21 @@ import org.apache.catalina.Context;
 import org.apache.catalina.startup.SimpleHttpClient;
 import org.apache.catalina.startup.Tomcat;
 import org.apache.catalina.startup.TomcatBaseTest;
+import org.apache.coyote.http11.Http11NioProtocol;
+import org.apache.coyote.http11.Http11Protocol;
 
 public class TestMaxConnections extends TomcatBaseTest{
 
-    Tomcat tomcat = null;
     static int soTimeout = 3000;
     static int connectTimeout = 1000;
     
-    public void test1() throws Exception {
-        init();
-        start();
+    @Override
+    public void setUp() throws Exception {
+        //do nothing
+    }
+    
+    public void testBio() throws Exception {
+        init(Http11Protocol.class.getName());
         ConnectThread[] t = new ConnectThread[10];
         int passcount = 0;
         int connectfail = 0;
@@ -52,54 +57,62 @@ public class TestMaxConnections extends TomcatBaseTest{
         }
         assertEquals("The number of successful requests should have been 5.",5, passcount);
         assertEquals("The number of failed connects should have been 5.",5, connectfail);
-        stop();
     }
     
+    public void testNio() throws Exception {
+        init(Http11NioProtocol.class.getName());
+        ConnectThread[] t = new ConnectThread[10];
+        int passcount = 0;
+        int connectfail = 0;
+        for (int i=0; i<t.length; i++) {
+            t[i] = new ConnectThread();
+            t[i].setName("ConnectThread["+i+"+]");
+            t[i].start();
+        }
+        for (int i=0; i<t.length; i++) {
+            t[i].join();
+            if (t[i].passed) passcount++;
+            if (t[i].connectfailed) connectfail++;
+        }
+        
+        assertTrue("The number of successful requests should have been 4-5, actual "+passcount,4==passcount || 5==passcount);
+        
+    }
 
     private class ConnectThread extends Thread {
         public boolean passed = true;
         public boolean connectfailed = false;
         public void run() {
             try {
-                TestKeepAliveClient client = new TestKeepAliveClient();
+                TestClient client = new TestClient();
                 client.doHttp10Request();
             }catch (Exception x) {
                 passed = false;
                 System.err.println(Thread.currentThread().getName()+" Error:"+x.getMessage());
-                connectfailed = "connect timed out".equals(x.getMessage());
+                connectfailed = "connect timed out".equals(x.getMessage()) || "Connection refused: connect".equals(x.getMessage());
             }
         }
     }
 
-    private boolean init;
     
-    private synchronized void init() {
-        if (init) return;
-        
+    private synchronized void init(String protocol) throws Exception {
+        System.setProperty("tomcat.test.protocol", protocol);
+        super.setUp();
         Tomcat tomcat = getTomcatInstance();
         Context root = tomcat.addContext("", SimpleHttpClient.TEMP_DIR);
         Tomcat.addServlet(root, "Simple", new SimpleServlet());
         root.addServletMapping("/test", "Simple");
-        tomcat.getConnector().setProperty("maxKeepAliveRequests", "5");
+        tomcat.getConnector().setProperty("maxKeepAliveRequests", "1");
         tomcat.getConnector().setProperty("soTimeout", "20000");
         tomcat.getConnector().setProperty("keepAliveTimeout", "50000");
         tomcat.getConnector().setProperty("port", "8080");
         tomcat.getConnector().setProperty("maxConnections", "4");
         tomcat.getConnector().setProperty("acceptCount", "1");
-        init = true;
-    }
-
-    private synchronized void start() throws Exception {
-        tomcat = getTomcatInstance();
-        init();
         tomcat.start();
+        Thread.sleep(5000);
     }
-    
-    private synchronized void stop() throws Exception {
-        tomcat.stop();
-    }
-    
-    private class TestKeepAliveClient extends SimpleHttpClient {
+
+    private class TestClient extends SimpleHttpClient {
 
         private void doHttp10Request() throws Exception {
             
@@ -123,35 +136,6 @@ public class TestMaxConnections extends TomcatBaseTest{
             assertTrue(passed);
         }
         
-        private void doHttp11Request() throws Exception {
-            Tomcat tomcat = getTomcatInstance();
-            init();
-            tomcat.start();
-            // Open connection
-            connect();
-            
-            // Send request in two parts
-            String[] request = new String[1];
-            request[0] =
-                "GET /test HTTP/1.1" + CRLF + 
-                "Host: localhost" + CRLF +
-                "Connection: Keep-Alive" + CRLF+
-                "Keep-Alive: 300"+ CRLF+ CRLF;
-            
-            setRequest(request);
-            
-            for (int i=0; i<5; i++) {
-                processRequest(false); // blocks until response has been read
-                assertTrue(getResponseLine()!=null && getResponseLine().trim().startsWith("HTTP/1.1 200"));
-            }
-            boolean passed = (this.readLine()==null);
-            // Close the connection
-            disconnect();
-            reset();
-            tomcat.stop();
-            assertTrue(passed);
-        }
-        
         @Override
         public boolean isResponseBodyOK() {
             return true;