Working

Author: robotdan

load-tests/self/src/main/java/io/fusionauth/http/load/LoadHandler.java

@@ -131,7 +131,6 @@ private void handleLoad(HTTPRequest req, HTTPResponse res) {
   }
 
   private void handleNoOp(HTTPRequest req, HTTPResponse res) {
-    // Note that it is intentionally that we are not reading the InputStream. This will cause the server to have to drain it.
     try (InputStream is = req.getInputStream()) {
       // Just read the bytes from the InputStream and return. Do no other worker.
       is.readAllBytes();

load-tests/self/src/main/java/io/fusionauth/http/load/Main.java

@@ -41,6 +41,7 @@ public static void main(String[] args) throws Exception {
 
     try (HTTPServer ignore = new HTTPServer().withHandler(new LoadHandler())
                                              .withCompressByDefault(false)
+                                             .withMaxRequestsPerConnection(100_000_000)
                                              .withInitialReadTimeout(Duration.ofSeconds(10))
                                              .withMinimumReadThroughput(4 * 1024)
                                              .withMinimumWriteThroughput(4 * 1024)

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

@@ -41,4 +41,5 @@ if [[ $# -ge 1 && $1 == "--suspend" ]]; then
   shift
 fi
 
+ulimit -S -n 32768
 ~/dev/java/current21/bin/java ${suspend} -cp "${CLASSPATH}" -Dio.fusionauth.http.server.stats="${SCRIPT_DIR}" io.fusionauth.http.load.Main

load-tests/tomcat/src/main/java/io/fusionauth/http/load/LoadServlet.java

@@ -137,7 +137,6 @@ private void handleLoad(HttpServletRequest req, HttpServletResponse res) {
   }
 
   private void handleNoOp(HttpServletRequest req, HttpServletResponse res) {
-    // Note that it is intentionally that we are not reading the InputStream. This will cause the server to have to drain it.
     try (InputStream is = req.getInputStream()) {
       // Just read the bytes from the InputStream and return. Do no other worker.
       is.readAllBytes();

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

@@ -107,6 +107,12 @@ public void run() {
           logger.trace("[{}] Accepted inbound connection. [{}] existing connections.", listenerAddress, clients.size());
         }
 
+        // TODO : Daniel : Review : Why is this number so much higher than the worker count when using persistent connections?
+        //        When using RESTIFY - we kill a lot of connections because read returns -1 while waiting for preamble.
+        //        This causes us to close a lot of workers. When using the JDK REST client this doesn't happen.
+        //        I don't know if this is just working as designed for HTTPURLConnection, or if it is related to
+        //        the read ahead we are doing.
+        //        Show Brian to see if he has any ideas.
         if (instrumenter != null) {
           instrumenter.acceptedConnection();
         }
@@ -187,8 +193,18 @@ public void run() {
           ClientInfo client = iterator.next();
           Thread thread = client.thread();
           long threadId = thread.threadId();
-          if (!thread.isAlive()) {
-            logger.trace("[{}] Remove dead client worker. Born [{}]. Died at age [{}] ms. Requests handled [{}].", threadId, client.getStartInstant(), client.getAge(), client.getHandledRequests());
+          boolean threadIsAlive = thread.isAlive();
+          var threadState = thread.getState();
+          if (threadState == Thread.State.TERMINATED) {
+            if (threadIsAlive) {
+              System.out.println("[" + threadId + "] is alive in state [" + threadState + "]");
+            }
+          } else if (!threadIsAlive) {
+            System.out.println("[" + threadId + "] is dead in state [" + threadState + "]");
+          }
+
+          if (!threadIsAlive) {
+            logger.debug("[{}] Remove dead client worker. Thread state [{}] Born [{}]. Died at age [{}] ms. Requests handled [{}].", threadId, threadState, client.getStartInstant(), client.getAge(), client.getHandledRequests());
             iterator.remove();
             removedClientCount++;
             continue;

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

@@ -22,8 +22,6 @@
 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;
@@ -34,6 +32,8 @@
 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;
@@ -69,8 +69,6 @@ public class HTTPWorker implements Runnable {
 
   private final Throughput throughput;
 
-  // TODO : Daniel : Review : Services such as Apache Tomcat have a maximum number of requests per keep-alive to protect against DOS attacks.
-  //        Needs more investigation, but we could cap the duration of worker by time or number of requests.
   private long handledRequests;
 
   private volatile State state;
@@ -130,6 +128,7 @@ public void run() {
         //        Daniel : Is there a risk of the preamble not needing the entire set of left over bytes and it trying to push
         //                  back bytes from it's own buffer? This would fail because we check that buffer is null when calling push.
         //                 But it would cause this request to fail.
+
         inputStream.setDelegate(new ThroughputInputStream(socket.getInputStream(), throughput));
 
         // Not this line of code will block
@@ -198,6 +197,7 @@ public void run() {
         if (!keepSocketAlive) {
           logger.trace("[{}] Closing socket. No Keep-Alive.", Thread.currentThread().threadId());
           closeSocketOnly(CloseSocketReason.Expected);
+          return;
         }
 
         // Transition to Keep-Alive state and reset the SO timeout
@@ -235,8 +235,8 @@ public void run() {
         logger.debug("[{}] Closing socket [{}]. {}.", Thread.currentThread().threadId(), state, message);
       }
       closeSocketOnly(reason);
-    } catch (ParseException pe) {
-      logger.info("[{}] Closing socket with status [{}]. Bad request, failed to parse request. Reason [{}] Parser state [{}]", Thread.currentThread().threadId(), Status.BadRequest, pe.getMessage(), pe.getState());
+    } catch (ParseException e) {
+      logger.debug("[{}] Closing socket with status [{}]. Bad request, failed to parse request. Reason [{}] Parser state [{}]", Thread.currentThread().threadId(), Status.BadRequest, e.getMessage(), e.getState());
       closeSocketOnError(response, Status.BadRequest);
     } catch (SocketException e) {
       // This should only happen when the server is shutdown and this thread is waiting to read or write. In that case, this will throw a
@@ -246,13 +246,13 @@ public void run() {
         logger.debug("[{}] Closing socket. Server is shutting down.", Thread.currentThread().threadId());
         closeSocketOnly(CloseSocketReason.Expected);
       }
-    } catch (IOException io) {
-      logger.debug(String.format("[%s] Closing socket with status [%d]. An IO exception was thrown during processing. These are pretty common.", Thread.currentThread().threadId(), Status.InternalServerError), io);
+    } catch (IOException e) {
+      logger.debug(String.format("[%s] Closing socket with status [%d]. An IO exception was thrown during processing. These are pretty common.", Thread.currentThread().threadId(), Status.InternalServerError), e);
       closeSocketOnError(response, Status.InternalServerError);
-    } catch (Throwable t) {
+    } catch (Throwable e) {
       // Log the error and signal a failure
       var status = Status.InternalServerError;
-      logger.error(String.format("[%s] Closing socket with status [%d]. An HTTP worker threw an exception while processing a request.", Thread.currentThread().threadId(), status), t);
+      logger.error(String.format("[%s] Closing socket with status [%d]. An HTTP worker threw an exception while processing a request.", Thread.currentThread().threadId(), status), e);
       closeSocketOnError(response, status);
     } finally {
       if (instrumenter != null) {