Working

Author: robotdan

load-tests/self/src/main/script/start.sh

@@ -36,7 +36,7 @@ for f in lib/*.jar; do
 done
 
 suspend=""
-if [[ $# -gte 1 && $1 == "--suspend" ]]; then
+if [[ $# -ge 1 && $1 == "--suspend" ]]; then
   suspend="-agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=8000"
   shift
 fi

src/main/java/io/fusionauth/http/io/BodyException.java renamed to src/main/java/io/fusionauth/http/BodyException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,7 +13,7 @@
  * either express or implied. See the License for the specific
  * language governing permissions and limitations under the License.
  */
-package io.fusionauth.http.io;
+package io.fusionauth.http;
 
 /**
  * Exception that is thrown if any HTTP body fails to be read and/or processed.

src/main/java/io/fusionauth/http/io/ChunkException.java renamed to src/main/java/io/fusionauth/http/ChunkException.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022-2024, FusionAuth, All Rights Reserved
+ * Copyright (c) 2022-2025, FusionAuth, All Rights Reserved
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -13,7 +13,7 @@
  * either express or implied. See the License for the specific
  * language governing permissions and limitations under the License.
  */
-package io.fusionauth.http.io;
+package io.fusionauth.http;
 
 /**
  * Exception that is thrown if a Chunked request or response is invalid.

src/main/java/io/fusionauth/http/ParseException.java

@@ -25,10 +25,6 @@ public class ParseException extends RuntimeException {
 
   private Integer index;
 
-  public ParseException() {
-    this.state = null;
-  }
-
   public ParseException(String message) {
     super(message);
     this.state = null;

src/main/java/io/fusionauth/http/io/ChunkedInputStream.java

@@ -18,6 +18,7 @@
 import java.io.IOException;
 import java.io.InputStream;
 
+import io.fusionauth.http.ChunkException;
 import io.fusionauth.http.ParseException;
 import io.fusionauth.http.util.HTTPTools;
 import static io.fusionauth.http.util.HTTPTools.makeParseException;
@@ -75,6 +76,13 @@ private int processChunk(byte[] destination, int offset, int length) throws IOEx
         int leftOver = bufferLength - bufferIndex;
         if (leftOver > 0) {
           // TODO : Daniel : Review : This doesn't seem like a good idea. It will fail silently, but this is required.
+          //        Discuss with Brian.
+          //        .
+          //        Options:
+          //         - Leave as is
+          //         - Throw a OverReadException that is caught by the HTTPInputStream which has a typed ref to PushbackInputStream and handled.
+          //         - Keep a typed ref for PushbackInputStream, but this messes up the 'commit' path in HTTPInputStream that swaps the pointer.
+          //                So we could re-work how we handle a chunked request body - instead of swapping out pointers we do something else?
           if (delegate instanceof PushbackInputStream pis) {
             pis.push(buffer, bufferIndex, leftOver);
           }
@@ -110,6 +118,7 @@ private int processChunk(byte[] destination, int offset, int length) throws IOEx
           int leftOver = bufferLength - bufferIndex;
           if (leftOver > 0) {
             // TODO : Daniel : Review : This doesn't seem like a good idea. It will fail silently, but this is required.
+            //        Discuss with Brian.
             if (delegate instanceof PushbackInputStream pis) {
               pis.push(buffer, bufferIndex, leftOver);
             }
@@ -155,6 +164,10 @@ private int processChunk(byte[] destination, int offset, int length) throws IOEx
 
         // TODO : If we wanted to handle more than one chunk at a time, I think we would potentially continue here
         //        and see if we can get more than one chunk completed before copying back to the destination.
+        //        Discuss with Brian.
+        //        Should we try this right now, or is this ok? Load testing with a small body,
+        //        Chunked is slower than fixed length, I suppose this makes sense. In practice I don't think browsers
+        //        and such use chunked for small requests.
 
         System.arraycopy(buffer, bufferIndex, destination, offset, lengthToCopy);
         bufferIndex += lengthToCopy;

src/main/java/io/fusionauth/http/server/Configurable.java

@@ -183,6 +183,18 @@ default T withMaxPendingSocketConnections(int maxPendingSocketConnections) {
     return (T) this;
   }
 
+  /**
+   * Sets the base directory for this server. This is passed to the HTTPContext, which is available from this class. This defaults to the
+   * current working directory of the process. Defaults to 100,000.
+   *
+   * @param maxRequestsPerConnection The maximum number of requests that can be handled by a single persistent connection.
+   * @return This.
+   */
+  default T withMaxRequestsPerConnection(int maxRequestsPerConnection) {
+    configuration().withMaxRequestsPerConnection(maxRequestsPerConnection);
+    return (T) this;
+  }
+
   /**
    * This configures the maximum size of a chunk in the response when the server is using chunked response encoding. Defaults to 16k.
    *

src/main/java/io/fusionauth/http/server/HTTPRequest.java

@@ -38,6 +38,7 @@
 import java.util.TreeSet;
 import java.util.stream.Collectors;
 
+import io.fusionauth.http.BodyException;
 import io.fusionauth.http.Buildable;
 import io.fusionauth.http.Cookie;
 import io.fusionauth.http.FileInfo;
@@ -47,7 +48,6 @@
 import io.fusionauth.http.HTTPValues.Headers;
 import io.fusionauth.http.HTTPValues.Protocols;
 import io.fusionauth.http.HTTPValues.TransferEncodings;
-import io.fusionauth.http.io.BodyException;
 import io.fusionauth.http.io.MultipartStream;
 import io.fusionauth.http.util.HTTPTools;
 import io.fusionauth.http.util.HTTPTools.HeaderValue;

src/main/java/io/fusionauth/http/server/HTTPResponse.java

@@ -285,6 +285,7 @@ public void reset() {
    * Reset the OutputStream.
    */
   // TODO : Daniel : Review : This is used by prime-mvc. Is there any other way to handle this?
+  //        Discuss with Brian. I think all prime-mvc needs is for the socket to be reset?
   public void resetOutputStream() {
     if (outputStream.isCommitted()) {
       throw new IllegalStateException("The HTTPResponse can't be reset after it has been committed, meaning at least one byte was written back to the client.");

src/main/java/io/fusionauth/http/server/HTTPServerConfiguration.java

@@ -52,9 +52,11 @@ public class HTTPServerConfiguration implements Configurable<HTTPServerConfigura
 
   private LoggerFactory loggerFactory = SystemOutLoggerFactory.FACTORY;
 
-  private int maxBytesToDrain = 128 * 1024; // 128k bytes
+  private int maxBytesToDrain = 256 * 1024; // 256k bytes
 
-  private int maxPendingSocketConnections = 200;
+  private int maxPendingSocketConnections = 250;
+
+  private int maxRequestsPerConnection = 100_000; // 100,000
 
   private int maxResponseChunkSize = 16 * 1024; // 16k bytes
 
@@ -68,9 +70,9 @@ public class HTTPServerConfiguration implements Configurable<HTTPServerConfigura
 
   private Duration readThroughputCalculationDelayDuration = Duration.ofSeconds(5);
 
-  private int requestBufferSize = 16 * 1024;
+  private int requestBufferSize = 16 * 1024; // 16k bytes
 
-  private int responseBufferSize = 64 * 1024;
+  private int responseBufferSize = 64 * 1024; // 16k bytes
 
   private Duration shutdownDuration = Duration.ofSeconds(10);
 
@@ -159,7 +161,7 @@ public LoggerFactory getLoggerFactory() {
   /**
    * @return The maximum number of bytes to drain from the InputStream when the request handler did not read all available bytes and the
    *     connection is using a keep-alive which means the server must drain the InputStream in preparation for the next request. Defaults to
-   *     128k.
+   *     256k.
    */
   public int getMaxBytesToDrain() {
     return maxBytesToDrain;
@@ -172,12 +174,22 @@ public int getMaxBytesToDrain() {
    * accepted by the server socket, a client socket is created and handed to an HTTP Worker. This queue length only needs to be large enough
    * to buffer the incoming requests as fast as we can accept them and hand them to a worker.
    *
-   * @return The maximum number of pending socket connections per HTTP listener. Defaults to 200.
+   * @return The maximum number of pending socket connections per HTTP listener. Defaults to 250.
    */
   public int getMaxPendingSocketConnections() {
     return maxPendingSocketConnections;
   }
 
+  /**
+   * This limit only applies when using a persistent connection. If this number is reached without hitting a Keep-Alive timeout the
+   * connection will be closed just as it would be if the Keep-Alive timeout was reached.
+   *
+   * @return The maximum number of requests that can be handled by a single persistent connection. Defaults to 100,000.
+   */
+  public int getMaxRequestsPerConnection() {
+    return maxRequestsPerConnection;
+  }
+
   /**
    * @return The max chunk size in the response. Defaults to 16k bytes.
    */
@@ -390,10 +402,27 @@ public HTTPServerConfiguration withLoggerFactory(LoggerFactory loggerFactory) {
    */
   @Override
   public HTTPServerConfiguration withMaxPendingSocketConnections(int maxPendingSocketConnections) {
+    if (maxPendingSocketConnections < 25) {
+      throw new IllegalArgumentException("The minimum pending socket connections must be greater than or equal to 25");
+    }
+
     this.maxPendingSocketConnections = maxPendingSocketConnections;
     return this;
   }
 
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public HTTPServerConfiguration withMaxRequestsPerConnection(int maxRequestsPerConnection) {
+    if (maxRequestsPerConnection < 10) {
+      throw new IllegalArgumentException("The maximum number of requests per connection must be greater than or equal to 10");
+    }
+
+    this.maxRequestsPerConnection = maxRequestsPerConnection;
+    return this;
+  }
+
   /**
    * {@inheritDoc}
    */

src/main/java/io/fusionauth/http/server/internal/HTTPWorker.java

@@ -22,6 +22,8 @@
 import java.net.SocketTimeoutException;
 
 import io.fusionauth.http.ConnectionClosedException;
+import io.fusionauth.http.server.HTTPRequest;
+import io.fusionauth.http.server.HTTPResponse;
 import io.fusionauth.http.HTTPValues;
 import io.fusionauth.http.HTTPValues.Connections;
 import io.fusionauth.http.HTTPValues.Headers;
@@ -32,8 +34,6 @@
 import io.fusionauth.http.log.Logger;
 import io.fusionauth.http.server.HTTPHandler;
 import io.fusionauth.http.server.HTTPListenerConfiguration;
-import io.fusionauth.http.server.HTTPRequest;
-import io.fusionauth.http.server.HTTPResponse;
 import io.fusionauth.http.server.HTTPServerConfiguration;
 import io.fusionauth.http.server.Instrumenter;
 import io.fusionauth.http.server.io.HTTPInputStream;
@@ -184,11 +184,14 @@ public void run() {
         logger.trace("[{}] Set state [{}]. Call the request handler.", Thread.currentThread().threadId(), state);
         configuration.getHandler().handle(request, response);
         logger.trace("[{}] Handler completed successfully", Thread.currentThread().threadId());
-        response.close();
 
-        // TODO : Daniel : Test
-        //                 Send a request body -> Handler ignores - does not read InputStream
-        //                 Handler, writes a large response. This should flush the OutputStream implicitly even w/out calling HTTPOutputStream.close()
+        // Do this before we write the response preamble. The normal Keep-Alive check below will handle closing the socket.
+        if (handledRequests >= configuration.getMaxRequestsPerConnection()) {
+          logger.trace("[{}] Maximum requests per connection has been reached. Turn off Keep-Alive.", Thread.currentThread().threadId());
+          response.setHeader(Headers.Connection, Connections.Close);
+        }
+
+        response.close();
 
         boolean keepSocketAlive = keepSocketAlive(request, response);
         // Close the socket.