Issue #14403 - Serialize FCGI input buffer access

jetty-core/jetty-fcgi/jetty-fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnection.java

@@ -38,6 +38,7 @@
 import org.eclipse.jetty.util.Attributes;
 import org.eclipse.jetty.util.Callback;
 import org.eclipse.jetty.util.StringUtil;
+import org.eclipse.jetty.util.thread.AutoLock;
 import org.eclipse.jetty.util.thread.ThreadPool;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -47,6 +48,7 @@ public class ServerFCGIConnection extends AbstractMetaDataConnection implements
     private static final Logger LOG = LoggerFactory.getLogger(ServerFCGIConnection.class);
 
     private final Callback fillableCallback = new FillableCallback();
+    private final AutoLock lock = new AutoLock();
     private final HttpChannel.Factory httpChannelFactory = new HttpChannel.DefaultFactory();
     private final Attributes attributes = new Lazy();
     private final Connector connector;
@@ -170,6 +172,14 @@ public void onOpen()
 
     @Override
     public void onFillable()
+    {
+        try (AutoLock ignored = lock.lock())
+        {
+            onFillableLocked();
+        }
+    }
+
+    private void onFillableLocked()
     {
         if (LOG.isDebugEnabled())
             LOG.debug(">>onFillable enter {} {} {}", this, stream, inputBuffer);
@@ -229,6 +239,14 @@ else if (read == 0)
      * for the current request.
      */
     void parseAndFill()
+    {
+        try (AutoLock ignored = lock.lock())
+        {
+            parseAndFillLocked();
+        }
+    }
+
+    private void parseAndFillLocked()
     {
         if (LOG.isDebugEnabled())
             LOG.debug("parseAndFill {}", this);
@@ -307,11 +325,14 @@ private void shutdown()
 
     void onCompleted(Throwable failure)
     {
-        releaseInputBuffer();
-        if (failure == null)
-            fillInterested(fillableCallback);
-        else
-            getFlusher().shutdown();
+        try (AutoLock ignored = lock.lock())
+        {
+            releaseInputBuffer();
+            if (failure == null)
+                fillInterested(fillableCallback);
+            else
+                getFlusher().shutdown();
+        }
     }
 
     @Override

jetty-core/jetty-fcgi/jetty-fcgi-server/src/test/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnectionTest.java

@@ -0,0 +1,123 @@
+//
+// ========================================================================
+// Copyright (c) 1995 Mort Bay Consulting Pty Ltd and others.
+//
+// This program and the accompanying materials are made available under the
+// terms of the Eclipse Public License v. 2.0 which is available at
+// https://www.eclipse.org/legal/epl-2.0, or the Apache License, Version 2.0
+// which is available at https://www.apache.org/licenses/LICENSE-2.0.
+//
+// SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
+// ========================================================================
+//
+
+package org.eclipse.jetty.fcgi.server.internal;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.eclipse.jetty.io.ByteArrayEndPoint;
+import org.eclipse.jetty.io.ByteBufferPool;
+import org.eclipse.jetty.io.RetainableByteBuffer;
+import org.eclipse.jetty.server.HttpConfiguration;
+import org.eclipse.jetty.server.Server;
+import org.eclipse.jetty.server.ServerConnector;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ServerFCGIConnectionTest
+{
+    @Test
+    public void testCompletionWaitsForFill() throws Exception
+    {
+        AtomicBoolean released = new AtomicBoolean();
+        AtomicBoolean releasedDuringFill = new AtomicBoolean();
+        ByteBufferPool bufferPool = new ByteBufferPool.Wrapper(ByteBufferPool.NON_POOLING)
+        {
+            @Override
+            public RetainableByteBuffer.Mutable acquire(int size, boolean direct)
+            {
+                return new RetainableByteBuffer.Mutable.Wrapper(super.acquire(size, direct))
+                {
+                    @Override
+                    public boolean release()
+                    {
+                        released.set(true);
+                        return super.release();
+                    }
+                };
+            }
+        };
+
+        CountDownLatch fillEntered = new CountDownLatch(1);
+        CountDownLatch fillProceed = new CountDownLatch(1);
+        ByteArrayEndPoint endPoint = new ByteArrayEndPoint()
+        {
+            @Override
+            public int fill(ByteBuffer buffer) throws IOException
+            {
+                fillEntered.countDown();
+                try
+                {
+                    assertTrue(fillProceed.await(5, TimeUnit.SECONDS));
+                }
+                catch (InterruptedException x)
+                {
+                    throw new IOException(x);
+                }
+                releasedDuringFill.set(released.get());
+                return super.fill(buffer);
+            }
+        };
+
+        Server server = new Server(null, null, bufferPool);
+        ServerConnector connector = new ServerConnector(server);
+        ServerFCGIConnection connection = new ServerFCGIConnection(connector, endPoint, new HttpConfiguration(), false);
+        AtomicReference<Throwable> fillFailure = new AtomicReference<>();
+        Thread fillThread = new Thread(() -> run(connection::onFillable, fillFailure));
+        fillThread.start();
+        assertTrue(fillEntered.await(5, TimeUnit.SECONDS));
+
+        AtomicReference<Throwable> completionFailure = new AtomicReference<>();
+        Thread completionThread = new Thread(() -> run(() -> connection.onCompleted(new IOException("test")), completionFailure));
+        completionThread.start();
+        await().atMost(5, TimeUnit.SECONDS).until(() -> !completionThread.isAlive() || completionThread.getState() == Thread.State.WAITING || completionThread.getState() == Thread.State.BLOCKED);
+
+        try
+        {
+            assertTrue(completionThread.isAlive());
+        }
+        finally
+        {
+            fillProceed.countDown();
+            fillThread.join(5000);
+            completionThread.join(5000);
+        }
+
+        assertFalse(fillThread.isAlive());
+        assertFalse(completionThread.isAlive());
+        assertFalse(releasedDuringFill.get());
+        assertNull(fillFailure.get());
+        assertNull(completionFailure.get());
+    }
+
+    private static void run(Runnable action, AtomicReference<Throwable> failure)
+    {
+        try
+        {
+            action.run();
+        }
+        catch (Throwable x)
+        {
+            failure.set(x);
+        }
+    }
+}