diff --git a/jetty-core/jetty-fcgi/jetty-fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnection.java b/jetty-core/jetty-fcgi/jetty-fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnection.java index b0460dd7c629..473e99b6f6d5 100644 --- a/jetty-core/jetty-fcgi/jetty-fcgi-server/src/main/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnection.java +++ b/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 diff --git a/jetty-core/jetty-fcgi/jetty-fcgi-server/src/test/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnectionTest.java b/jetty-core/jetty-fcgi/jetty-fcgi-server/src/test/java/org/eclipse/jetty/fcgi/server/internal/ServerFCGIConnectionTest.java new file mode 100644 index 000000000000..5d7b012a80ef --- /dev/null +++ b/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 fillFailure = new AtomicReference<>(); + Thread fillThread = new Thread(() -> run(connection::onFillable, fillFailure)); + fillThread.start(); + assertTrue(fillEntered.await(5, TimeUnit.SECONDS)); + + AtomicReference 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 failure) + { + try + { + action.run(); + } + catch (Throwable x) + { + failure.set(x); + } + } +}