Fix Jetty 12 active request accounting

Track each HTTP request independently and always release its entry on
completion, failure, or async termination. Enforce the request limit
atomically, remove obsolete connection-close accounting, repair stale
cleanup, and persist the configured maximum correctly.

Update UI terminology, help, localization, and focused regression tests.

- Fixes #800
- Fixes #808

source/net/yacy/cora/protocol/ConnectionInfo.java

@@ -44,16 +44,17 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
             .synchronizedSet(new HashSet<ConnectionInfo>());
     private final static Set<ConnectionInfo> serverConnections = Collections
             .synchronizedSet(new HashSet<ConnectionInfo>());
-    // this is only for statistics, so it can be bigger to see lost connectionInfos
-    private final static int staleAfterMillis = 30 * 60000; // 30 minutes
+    /* Stale cleanup is a safety net; normal request and client lifecycles remove their own entries. */
+    private static final long CLIENT_STALE_AFTER_MILLIS = 30L * 60L * 1000L;
+    private static final long SERVER_STALE_AFTER_MILLIS = 30L * 60L * 1000L;
     
     private static int maxcount = 20;
-    private static int serverMaxCount = 50;
+    private static volatile int serverMaxCount = 50;
 
     private final String protocol;
     private final String targetHost;
     private final String command;
-    private final int id;
+    private final long id;
     private final long initTime;
     private final long upbytes;
 
@@ -66,7 +67,7 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
      * @param id
      * @param initTime
      */
-    public ConnectionInfo(final String protocol, final String targetHost, final String command, final int id,
+    public ConnectionInfo(final String protocol, final String targetHost, final String command, final long id,
             final long initTime, final long upbytes) {
         this.protocol = protocol;
         this.targetHost = targetHost;
@@ -114,7 +115,7 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
     /**
      * @return
      */
-    public int getID() {
+    public long getID() {
         return id;
     }
 
@@ -150,7 +151,7 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
     }
     
     /**
-     * gets the number of active server connections
+     * gets the number of active incoming server requests
      * 
      * @return count of active connections
      */
@@ -168,7 +169,7 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
     }
     
     /**
-     * @return wether the server max connection-count is reached
+     * @return whether the incoming server request limit is reached
      */
     public static boolean isServerCountReached() {
     	return getServerCount() >= getServerMaxcount();
@@ -210,7 +211,7 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
     }
     
     /**
-     * gets the max connection count of the Server connection manager
+     * gets the maximum active incoming server request count
      * 
      * @return max connections
      */
@@ -219,14 +220,17 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
     }
     
     /**
-     * gets the max connection count of the Sever connection manager
-     * to be used in statistics
+     * sets the maximum active incoming server request count
      * 
      * @param max connections
      * @TODO Is it correct to only set if max > 0? What if maxcount is > 0 and max = 0 ?
      */
     public static void setServerMaxcount(final int max) {
-    	if (max > 0) serverMaxCount = max;
+        if (max > 0) {
+            synchronized (serverConnections) {
+                serverMaxCount = max;
+            }
+        }
     }
 
     /**
@@ -239,12 +243,30 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
     }
 
     /**
-     * add a Server connection to the list of all current connections
+     * add an incoming server request to the active request list
      * 
      * @param conInfo
      */
     public static void addServerConnection(final ConnectionInfo conInfo) {
-    	getServerConnections().add(conInfo);
+        getServerConnections().add(conInfo);
+    }
+
+    /**
+     * Add an incoming server request when capacity is available. The capacity check
+     * and insertion are performed while holding the same lock, so concurrent
+     * callers cannot exceed the configured limit.
+     *
+     * @param conInfo the connection to track
+     * @param limitExempt when true, add the connection regardless of the limit
+     * @return true when the connection was added, false when it was rejected
+     */
+    public static boolean tryAddServerConnection(final ConnectionInfo conInfo, final boolean limitExempt) {
+        synchronized (serverConnections) {
+            if (!limitExempt && serverConnections.size() >= serverMaxCount) {
+                return false;
+            }
+            return serverConnections.add(conInfo);
+        }
     }
 
     /**
@@ -257,7 +279,7 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
     }
 
     /**
-     * remove a Server connection from the list of all current connections
+     * remove an incoming server request from the active request list
      * 
      * @param conInfo
      */
@@ -270,42 +292,50 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
      * 
      * @param id
      */
-    public static void removeConnection(final int id) {
+    public static void removeConnection(final long id) {
         removeConnection(new ConnectionInfo(null, null, null, id, 0, 0));
     }
 
     /**
-     * connections with same id {@link equals()} another
-     * 
-     * @param id
-     */
-    public static void removeServerConnection(final int id) {
-        removeServerConnection(new ConnectionInfo(null, null, null, id, 0, 0));
-    }
-    
-    /**
-     * removes stale connections
+     * Remove stale client and server entries. Prefer the pool-specific cleanup
+     * methods when the caller treats the two pools independently.
      */
     public static void cleanUp() {
-    	cleanup(getAllConnections());
-    	cleanup(getServerConnections());
+        cleanUpClientConnections();
+        cleanUpServerConnections();
     }
-    
-    private static void cleanup(final Set<ConnectionInfo> connectionSet) {
-    	final Iterator<ConnectionInfo> iter = connectionSet.iterator();
-    	synchronized (iter) { 
+
+    /**
+     * Remove client connection entries that outlived the stale threshold.
+     *
+     * @return the number of removed entries
+     */
+    public static int cleanUpClientConnections() {
+        return cleanup(allConnections, CLIENT_STALE_AFTER_MILLIS);
+    }
+
+    /**
+     * Remove incoming request entries that outlived the stale threshold.
+     *
+     * @return the number of removed entries
+     */
+    public static int cleanUpServerConnections() {
+        return cleanup(serverConnections, SERVER_STALE_AFTER_MILLIS);
+    }
+
+    private static int cleanup(final Set<ConnectionInfo> connectionSet, final long staleAfterMillis) {
+        int removed = 0;
+        synchronized (connectionSet) {
+            final Iterator<ConnectionInfo> iter = connectionSet.iterator();
             while (iter.hasNext()) {
-                ConnectionInfo con = null;
-                try {
-                    con = iter.next();
-                } catch (Throwable e) {break;} // this must break because otherwise there is danger that the loop does never terminates
-                try {
-                    if (con.getLifetime() > staleAfterMillis) {
-                        connectionSet.remove(con);
-                    }
-                } catch (Throwable e) {continue;}
+                final ConnectionInfo con = iter.next();
+                if (con.getLifetime() > staleAfterMillis) {
+                    iter.remove();
+                    removed++;
+                }
             }
         }
+        return removed;
     }
     
     /* (non-Javadoc)
@@ -335,10 +365,7 @@ public class ConnectionInfo implements Comparable<ConnectionInfo> {
      */
     @Override
     public int hashCode() {
-        final int prime = 31;
-        int result = 1;
-        result = prime * result + id;
-        return result;
+        return Long.hashCode(this.id);
     }
 
     /*

source/net/yacy/htroot/Connections_p.java

@@ -44,7 +44,7 @@ public final class Connections_p {
         // return variable that accumulates replacements
         final serverObjects prop = new serverObjects();
 
-        // server sessions
+        // active incoming HTTP requests
         List<ConnectionInfo> allConnectionsSorted = new LinkedList<ConnectionInfo>(ConnectionInfo.getServerConnections());
         Collections.sort(allConnectionsSorted);
         Collections.reverse(allConnectionsSorted); // toggle ascending/descending
@@ -64,7 +64,7 @@ public final class Connections_p {
         prop.putNum("numMax", ConnectionInfo.getServerMaxcount());
         prop.putNum("numActiveRunning", c);
 
-        // client sessions
+        // active outgoing client connections
         // sorting: sort by initTime, decending
         allConnectionsSorted = new LinkedList<ConnectionInfo>(ConnectionInfo.getAllConnections());
         Collections.sort(allConnectionsSorted);

source/net/yacy/htroot/PerformanceQueues_p.java

@@ -52,6 +52,12 @@ import net.yacy.server.serverSwitch;
 
 public class PerformanceQueues_p {
 
+    static final String INCOMING_HTTP_REQUESTS_POOL_NAME = "Incoming HTTP Requests";
+    static final String INCOMING_HTTP_REQUESTS_MAX_ACTIVE_PARAM =
+            INCOMING_HTTP_REQUESTS_POOL_NAME + "_maxActive";
+    static final String LEGACY_HTTPD_SESSION_POOL_MAX_ACTIVE_PARAM =
+            "httpd Session Pool_maxActive";
+
     @SuppressWarnings("deprecation")
 	public static serverObjects respond(final RequestHeader header, final serverObjects post, final serverSwitch env) {
         // return variable that accumulates replacements
@@ -302,18 +308,9 @@ public class PerformanceQueues_p {
             sb.setConfig(SwitchboardConstants.ROBOTS_TXT_THREADS_ACTIVE_MAX, maxBusy);
 
             /*
-             * configuring the http pool
+             * configuring the incoming HTTP request limit
              */
-            try {
-                maxBusy = post.getInt("httpd Session Pool_maxActive", 8);
-            } catch (final NumberFormatException e) {
-                maxBusy = 8;
-            }
-
-            ConnectionInfo.setServerMaxcount(maxBusy);
-
-            // storing the new values into configfile
-            sb.setConfig("httpdMaxBusySessions",maxBusy);
+            applyServerMaxcount(sb, postedServerMaxcount(post));
 
         }
 
@@ -357,16 +354,19 @@ public class PerformanceQueues_p {
         prop.putNum("crawlPauseLocalsearchCurrent", (System.currentTimeMillis() - sb.localSearchLastAccess) / 1000);
         prop.putNum("crawlPauseRemotesearchCurrent", (System.currentTimeMillis() - sb.remoteSearchLastAccess) / 1000);
 
-        // table thread pool settings
+        // concurrency limits
         prop.put("pool_0_name","Crawler Pool");
+        prop.put("pool_0_incomingRequests", 0);
         prop.put("pool_0_maxActive", sb.getConfigLong(SwitchboardConstants.CRAWLER_THREADS_ACTIVE_MAX, 0));
         prop.put("pool_0_numActive", sb.crawlQueues.activeWorkerEntries().size());
 
         prop.put("pool_1_name","Robots.txt Pool");
+        prop.put("pool_1_incomingRequests", 0);
         prop.put("pool_1_maxActive", sb.getConfigInt(SwitchboardConstants.ROBOTS_TXT_THREADS_ACTIVE_MAX, SwitchboardConstants.ROBOTS_TXT_THREADS_ACTIVE_MAX_DEFAULT));
         prop.put("pool_1_numActive", sb.crawlQueues.activeWorkerEntries().size());
 
-        prop.put("pool_2_name", "httpd Session Pool");
+        prop.put("pool_2_name", INCOMING_HTTP_REQUESTS_POOL_NAME);
+        prop.put("pool_2_incomingRequests", 1);
         prop.put("pool_2_maxActive", ConnectionInfo.getServerMaxcount());
         prop.put("pool_2_numActive", ConnectionInfo.getServerCount());
 
@@ -420,6 +420,28 @@ public class PerformanceQueues_p {
         return prop;
     }
 
+    /** Apply and persist an incoming request limit only when it is valid. */
+    static boolean applyServerMaxcount(final serverSwitch env, final int maxBusy) {
+        if (maxBusy <= 0) {
+            return false;
+        }
+        ConnectionInfo.setServerMaxcount(maxBusy);
+        env.setConfig(SwitchboardConstants.SERVER_MAX_BUSY_SESSIONS, maxBusy);
+        return true;
+    }
+
+    /** Read the corrected form parameter, with the former session-pool name as an alias. */
+    static int postedServerMaxcount(final serverObjects post) {
+        final String parameter = post.containsKey(INCOMING_HTTP_REQUESTS_MAX_ACTIVE_PARAM)
+                ? INCOMING_HTTP_REQUESTS_MAX_ACTIVE_PARAM
+                : LEGACY_HTTPD_SESSION_POOL_MAX_ACTIVE_PARAM;
+        try {
+            return post.getInt(parameter, ConnectionInfo.getServerMaxcount());
+        } catch (final NumberFormatException e) {
+            return ConnectionInfo.getServerMaxcount();
+        }
+    }
+
     /**
      * Parse a JVM maximum heap size option like "Xmx600m", "Xmx16g" or "Xmx2048k"
      * into mebibytes for display and editing on the Performance page. The value may

source/net/yacy/http/Jetty12HttpServer.java

@@ -23,7 +23,6 @@ package net.yacy.http;
 import java.io.IOException;
 import java.io.Writer;
 import java.net.InetAddress;
-import java.net.InetSocketAddress;
 import java.nio.channels.ClosedChannelException;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -47,7 +46,6 @@ import org.eclipse.jetty.ee8.servlet.ServletHolder;
 import org.eclipse.jetty.ee8.webapp.WebAppContext;
 import org.eclipse.jetty.http.HttpMethod;
 import org.eclipse.jetty.http.HttpVersion;
-import org.eclipse.jetty.io.Connection;
 import org.eclipse.jetty.io.EofException;
 import org.eclipse.jetty.security.HashLoginService;
 import org.eclipse.jetty.security.UserPrincipal;
@@ -70,7 +68,6 @@ import org.eclipse.jetty.util.ssl.SslContextFactory;
 
 import com.google.common.net.InetAddresses;
 
-import net.yacy.cora.protocol.ConnectionInfo;
 import net.yacy.cora.protocol.RequestHeader;
 import net.yacy.cora.util.ConcurrentLog;
 import net.yacy.http.servlets.MonitorFilter;
@@ -123,24 +120,12 @@ public class Jetty12HttpServer implements YaCyHttpServer {
     private static Server createServer(final int port, final String host, final int acceptorCount,
             final SSLContext sslContext, final int sslPort) {
         final Server server = new Server();
-        final Connection.Listener connectionCloseMonitor = new Connection.Listener() {
-            @Override
-            public void onClosed(final Connection connection) {
-                if (connection.getEndPoint().getRemoteSocketAddress() instanceof InetSocketAddress) {
-                    final InetSocketAddress remote =
-                            (InetSocketAddress) connection.getEndPoint().getRemoteSocketAddress();
-                    ConnectionInfo.removeServerConnection(MonitorFilter.connectionId(
-                            remote.getAddress().getHostAddress(), remote.getPort()));
-                }
-            }
-        };
 
         final HttpConfiguration httpConfiguration = new HttpConfiguration();
         httpConfiguration.setRequestHeaderSize(HttpServerBootstrapConfig.REQUEST_HEADER_SIZE);
         final ServerConnector httpConnector = new ServerConnector(server, acceptorCount, -1,
                 new HttpConnectionFactory(httpConfiguration));
-        configureConnector(httpConnector, host, port, "httpd-" + host + ":" + port,
-                connectionCloseMonitor);
+        configureConnector(httpConnector, host, port, "httpd-" + host + ":" + port);
         httpConnector.setAcceptQueueSize(HttpServerBootstrapConfig.ACCEPT_QUEUE_SIZE);
         server.addConnector(httpConnector);
 
@@ -152,8 +137,7 @@ public class Jetty12HttpServer implements YaCyHttpServer {
             final ServerConnector httpsConnector = new ServerConnector(server, acceptorCount, -1,
                     new SslConnectionFactory(sslContextFactory, HttpVersion.HTTP_1_1.asString()),
                     new HttpConnectionFactory(httpsConfiguration));
-            configureConnector(httpsConnector, host, sslPort, "ssld:" + sslPort,
-                    connectionCloseMonitor);
+            configureConnector(httpsConnector, host, sslPort, "ssld:" + sslPort);
             httpsConnector.setAcceptQueueSize(HttpServerBootstrapConfig.ACCEPT_QUEUE_SIZE);
             server.addConnector(httpsConnector);
         }
@@ -169,12 +153,11 @@ public class Jetty12HttpServer implements YaCyHttpServer {
     }
 
     private static void configureConnector(final ServerConnector connector, final String host,
-            final int port, final String name, final Connection.Listener listener) {
+            final int port, final String name) {
         connector.setHost(host);
         connector.setPort(port);
         connector.setName(name);
         connector.setIdleTimeout(HttpServerBootstrapConfig.CONNECTOR_IDLE_TIMEOUT_MILLIS);
-        connector.addBean(listener);
     }
 
     private static Handler createWebAppHandler(final Server server,

source/net/yacy/http/servlets/MonitorFilter.java

@@ -25,7 +25,11 @@
 package net.yacy.http.servlets;
 
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 
+import javax.servlet.AsyncEvent;
+import javax.servlet.AsyncListener;
 import javax.servlet.Filter;
 import javax.servlet.FilterChain;
 import javax.servlet.FilterConfig;
@@ -42,14 +46,17 @@ import net.yacy.cora.protocol.RequestHeader;
 /**
  * Records incoming server requests into {@link ConnectionInfo} (displayed on
  * Connections_p.html) and rejects remote requests with http status 503 when
- * the configured maximum number of server connections is reached.
+ * the configured maximum number of active incoming requests is reached.
  *
- * This is a plain servlet filter (former Jetty handler MonitorHandler); the
- * tracking entry of a connection is removed on connection close by a
- * servlet-container specific listener, see Jetty12HttpServer.
+ * This is a plain servlet filter (former Jetty handler MonitorHandler). A
+ * synchronous request is removed when its filter chain exits; an asynchronous
+ * request is removed by a listener when its asynchronous lifecycle finishes.
  */
 public class MonitorFilter implements Filter {
 
+    /** Supplies a distinct tracking identity for every incoming request. */
+    private static final AtomicLong NEXT_REQUEST_ID = new AtomicLong();
+
     @Override
     public void init(FilterConfig filterConfig) throws ServletException {
     }
@@ -58,18 +65,6 @@ public class MonitorFilter implements Filter {
     public void destroy() {
     }
 
-    /**
-     * @return the ConnectionInfo id of the tcp connection identified by client address and port
-     */
-    public static int connectionId(String remoteAddr, final int remotePort) {
-        // the servlet API reports IPv6 addresses in bracketed form ("[::1]"), the
-        // connection endpoint in plain form: normalize to the plain form
-        if (remoteAddr.startsWith("[") && remoteAddr.endsWith("]")) {
-            remoteAddr = remoteAddr.substring(1, remoteAddr.length() - 1);
-        }
-        return (remoteAddr + ":" + remotePort).hashCode();
-    }
-
     @Override
     public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
             throws IOException, ServletException {
@@ -80,20 +75,56 @@ public class MonitorFilter implements Filter {
                 hrequest.getScheme(),
                 RequestHeader.client(hrequest) + ":" + hrequest.getRemotePort(),
                 hrequest.getMethod() + " " + hrequest.getRequestURI() + (query == null ? "" : "?" + query),
-                connectionId(hrequest.getRemoteAddr(), hrequest.getRemotePort()),
+                NEXT_REQUEST_ID.getAndIncrement(),
                 System.currentTimeMillis(),
                 -1);
 
-        // a keep-alive connection reuses the id: remove a previous entry to show the latest request
-        ConnectionInfo.removeServerConnection(info);
-        ConnectionInfo.addServerConnection(info);
-
-        if (ConnectionInfo.isServerCountReached()
-                && !Domains.isLocal(hrequest.getRemoteAddr(), null)) {
+        final boolean limitExempt = Domains.isLocal(hrequest.getRemoteAddr(), null);
+        if (!ConnectionInfo.tryAddServerConnection(info, limitExempt)) {
             ((HttpServletResponse) response).sendError(HttpServletResponse.SC_SERVICE_UNAVAILABLE,
-                    "max. server connections reached (increase /PerformanceQueues_p.html -> httpd Session Pool).");
+                    "maximum active HTTP requests reached (increase Incoming HTTP Requests on "
+                            + "/PerformanceQueues_p.html).");
             return;
         }
-        chain.doFilter(request, response);
+
+        final AtomicBoolean removed = new AtomicBoolean();
+        final Runnable removeTracking = () -> {
+            if (removed.compareAndSet(false, true)) {
+                ConnectionInfo.removeServerConnection(info);
+            }
+        };
+        boolean asyncCleanupRegistered = false;
+        try {
+            chain.doFilter(request, response);
+
+            if (hrequest.isAsyncStarted()) {
+                hrequest.getAsyncContext().addListener(new AsyncListener() {
+                    @Override
+                    public void onComplete(final AsyncEvent event) {
+                        removeTracking.run();
+                    }
+
+                    @Override
+                    public void onTimeout(final AsyncEvent event) {
+                        removeTracking.run();
+                    }
+
+                    @Override
+                    public void onError(final AsyncEvent event) {
+                        removeTracking.run();
+                    }
+
+                    @Override
+                    public void onStartAsync(final AsyncEvent event) {
+                        event.getAsyncContext().addListener(this);
+                    }
+                });
+                asyncCleanupRegistered = true;
+            }
+        } finally {
+            if (!asyncCleanupRegistered) {
+                removeTracking.run();
+            }
+        }
     }
 }

source/net/yacy/search/Switchboard.java

@@ -2547,8 +2547,9 @@ public final class Switchboard extends serverSwitch {
                 ConcurrentLog.logException(e);
             }
 
-            // close unused connections
-            ConnectionInfo.cleanUp();
+            // Safety-net cleanup for entries missed by their normal lifecycle handling.
+            ConnectionInfo.cleanUpClientConnections();
+            ConnectionInfo.cleanUpServerConnections();
 
             // clean up delegated stack
             this.checkInterruption();

source/net/yacy/search/SwitchboardConstants.java

@@ -66,6 +66,9 @@ public final class SwitchboardConstants {
     public static final String SERVER_SHUTDOWNPORT          = "port.shutdown"; // local port to listen for a shutdown signal (0 <= disabled)
     public static final String SERVER_STATICIP              = "staticIP"; // static IP of http server
     public static final String SERVER_PUBLICPORT            = "publicPort";
+    /** Maximum number of concurrently tracked incoming HTTP requests. */
+    public static final String SERVER_MAX_BUSY_SESSIONS     = "httpdMaxBusySessions";
+    public static final int SERVER_MAX_BUSY_SESSIONS_DEFAULT = 200;
     /** Socket peers whose validated X-Real-IP header may be used for routing and tracking. */
     public static final String SERVER_REVERSE_PROXY_TRUSTED = "server.reverseProxy.trusted";
     public static final String SERVER_REVERSE_PROXY_TRUSTED_DEFAULT =

source/net/yacy/yacy.java

@@ -298,15 +298,16 @@ public final class yacy {
             final String host = sb.getLocalHost();
             ScheduledExecutorService logReportScheduler = null;
             try {
-                // start http server
-                YaCyHttpServer httpServer;
-                httpServer = new Jetty12HttpServer(port, host);
-                httpServer.startupServer();
-                sb.setHttpServer(httpServer);
-                // TODO: this has no effect on Jetty (but needed to reflect configured value and limit is still used)
-                ConnectionInfo.setServerMaxcount(sb.getConfigInt("connectionsMax", ConnectionInfo.getMaxcount()));
-
-                ConcurrentLog.info("STARTUP",httpServer.getVersion());
+                // start http server
+                YaCyHttpServer httpServer;
+                ConnectionInfo.setServerMaxcount(sb.getConfigInt(
+                        SwitchboardConstants.SERVER_MAX_BUSY_SESSIONS,
+                        SwitchboardConstants.SERVER_MAX_BUSY_SESSIONS_DEFAULT));
+                httpServer = new Jetty12HttpServer(port, host);
+                httpServer.startupServer();
+                sb.setHttpServer(httpServer);
+
+                ConcurrentLog.info("STARTUP",httpServer.getVersion());
 
                 // open the browser window
                 final boolean browserPopUpTrigger = sb.getConfig(SwitchboardConstants.BROWSER_POP_UP_TRIGGER, "true").equals("true");