Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Undisable ee9 BlockingTest and fix HttpChannel.produceContent #12529

Merged
merged 13 commits into from
Dec 18, 2024
Merged

Undisable ee9 BlockingTest and fix HttpChannel.produceContent #12529

Show file tree
Hide file tree
Changes from 9 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
e510b69
Undisable ee9 BlockingTest and fix HttpChannel.produceContent
janbartel Nov 13, 2024
222a272
Handle CoreContextRequest being null in produceContent/needContent
lorban Nov 25, 2024
546cdd9
make tests non-flaky
lorban Dec 9, 2024
debb110
fix BlockingTest.testBlockingReadAndBlockingWriteGzipped hanging b/c …
lorban Dec 9, 2024
fe418fd
handle review comments
lorban Dec 10, 2024
1a8c6f2
handle review comments
lorban Dec 10, 2024
8e33d28
handle review comments
lorban Dec 10, 2024
f90f141
add temporary debugging stack trace
lorban Dec 10, 2024
673391b
fix exception type
lorban Dec 10, 2024
c86a935
fix exception type
lorban Dec 11, 2024
bc191b5
remove leftover
lorban Dec 11, 2024
4a149e9
do not recycle HttpStreamOverHttp1 when failed so that ChannelRequest…
lorban Dec 11, 2024
9cfe2de
restore breaking the link between channel and stream
lorban Dec 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 22 additions & 6 deletions ...y-core/jetty-server/src/main/java/org/eclipse/jetty/server/internal/HttpChannelState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1318,11 +1318,19 @@ public void succeeded()
{
callback = _writeCallback;
_writeCallback = null;
httpChannel = _request.lockedGetHttpChannelState();
httpChannel.lockedStreamSendCompleted(true);
httpChannel = _request._httpChannelState;
lorban marked this conversation as resolved.
Show resolved Hide resolved
if (httpChannel != null)
httpChannel.lockedStreamSendCompleted(true);
}

if (callback != null)
httpChannel._writeInvoker.run(callback::succeeded);
{
if (httpChannel != null)
httpChannel._writeInvoker.run(callback::succeeded);
else
// Channel already completed, just fail the callback.
HttpChannelState.failed(callback, new IllegalStateException("channel already completed"));
lorban marked this conversation as resolved.
Show resolved Hide resolved
}
}

/**
Expand All @@ -1347,11 +1355,19 @@ public void failed(Throwable x)
_writeFailure = x;
callback = _writeCallback;
_writeCallback = null;
httpChannel = _request.lockedGetHttpChannelState();
httpChannel.lockedStreamSendCompleted(false);
httpChannel = _request._httpChannelState;
if (httpChannel != null)
httpChannel.lockedStreamSendCompleted(false);
}

if (callback != null)
httpChannel._writeInvoker.run(() -> HttpChannelState.failed(callback, x));
{
if (httpChannel != null)
httpChannel._writeInvoker.run(() -> HttpChannelState.failed(callback, x));
else
// Channel already completed, just fail the callback.
HttpChannelState.failed(callback, ExceptionUtil.combine(x, new IllegalStateException("channel already completed")));
lorban marked this conversation as resolved.
Show resolved Hide resolved
}
}

@Override
Expand Down
5 changes: 5 additions & 0 deletions jetty-ee8/jetty-ee8-nested/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,11 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-http-tools</artifactId>
Expand Down
5 changes: 5 additions & 0 deletions jetty-ee9/jetty-ee9-nested/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,11 @@
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.eclipse.jetty</groupId>
<artifactId>jetty-http-tools</artifactId>
Expand Down
11 changes: 9 additions & 2 deletions jetty-ee9/jetty-ee9-nested/src/main/java/org/eclipse/jetty/ee9/nested/HttpChannel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -156,8 +156,12 @@ public ConnectionMetaData getConnectionMetaData()
*/
public boolean needContent()
{
ContextHandler.CoreContextRequest coreContextRequest = getCoreRequest();
// When coreContextRequest is null, produceContent() always immediately returns an error content.
if (coreContextRequest == null)
return true;
// TODO: optimize by attempting a read?
getCoreRequest().demand(_needContentTask);
coreContextRequest.demand(_needContentTask);
return false;
}

Expand All @@ -172,7 +176,10 @@ public boolean needContent()
*/
public HttpInput.Content produceContent()
{
Content.Chunk chunk = getCoreRequest().read();
ContextHandler.CoreContextRequest coreContextRequest = getCoreRequest();
if (coreContextRequest == null)
return new HttpInput.ErrorContent(new IOException("Channel has been recycled"));
Content.Chunk chunk = coreContextRequest.read();
if (chunk == null)
return null;

Expand Down
85 changes: 57 additions & 28 deletions jetty-ee9/jetty-ee9-nested/src/test/java/org/eclipse/jetty/ee9/nested/BlockingTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,15 +33,16 @@
import jakarta.servlet.ServletOutputStream;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import org.awaitility.Awaitility;
import org.eclipse.jetty.http.HttpTester;
import org.eclipse.jetty.server.Server;
import org.eclipse.jetty.server.ServerConnector;
import org.eclipse.jetty.server.handler.gzip.GzipHandler;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.Test;

import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.instanceOf;
Expand All @@ -51,7 +52,6 @@
import static org.hamcrest.core.Is.is;
import static org.junit.jupiter.api.Assertions.assertTrue;

@Disabled // TODO
public class BlockingTest
{
private Server server;
Expand Down Expand Up @@ -86,7 +86,7 @@ public void testBlockingReadThenNormalComplete() throws Exception
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
new Thread(() ->
Thread thread = new Thread(() ->
{
try
{
Expand All @@ -103,14 +103,19 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
readException.set(t);
stopped.countDown();
}
}).start();
});
thread.start();

try
{
// wait for thread to start and read first byte
started.await(10, TimeUnit.SECONDS);
assertTrue(started.await(10, TimeUnit.SECONDS));
// give it time to block on second byte
Thread.sleep(1000);
await().atMost(5, TimeUnit.SECONDS).until(() ->
{
Thread.State state = thread.getState();
return state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING;
});
}
catch (Throwable e)
{
Expand Down Expand Up @@ -249,26 +254,28 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
@Test
public void testNormalCompleteThenBlockingRead() throws Exception
{
Awaitility.setDefaultPollInterval(1, TimeUnit.MICROSECONDS);
lorban marked this conversation as resolved.
Show resolved Hide resolved

CountDownLatch started = new CountDownLatch(1);
CountDownLatch completed = new CountDownLatch(1);
CountDownLatch stopped = new CountDownLatch(1);
AtomicReference<Throwable> readException = new AtomicReference<>();
AtomicReference<Thread> threadRef = new AtomicReference<>();
AbstractHandler handler = new AbstractHandler()
{
@Override
public void handle(String target, Request baseRequest, HttpServletRequest request, HttpServletResponse response) throws IOException, ServletException
{
baseRequest.setHandled(true);
new Thread(() ->
Thread thread = new Thread(() ->
{
try
{
int b = baseRequest.getHttpInput().read();
if (b == '1')
{
started.countDown();
completed.await(10, TimeUnit.SECONDS);
Thread.sleep(500);
assertTrue(completed.await(10, TimeUnit.SECONDS));
if (baseRequest.getHttpInput().read() > Integer.MIN_VALUE)
throw new IllegalStateException();
}
Expand All @@ -278,14 +285,20 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
readException.set(t);
stopped.countDown();
}
}).start();
});
threadRef.set(thread);
thread.start();

try
{
// wait for thread to start and read first byte
started.await(10, TimeUnit.SECONDS);
assertTrue(started.await(10, TimeUnit.SECONDS));
// give it time to block on second byte
Thread.sleep(1000);
await().atMost(5, TimeUnit.SECONDS).until(() ->
{
Thread.State state = thread.getState();
return state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING;
});
}
catch (Throwable e)
{
Expand Down Expand Up @@ -321,7 +334,7 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
assertThat(response.getContent(), containsString("OK"));

completed.countDown();
Thread.sleep(1000);
await().atMost(5, TimeUnit.SECONDS).until(() -> threadRef.get().getState() == Thread.State.TERMINATED);

// Async thread should have stopped
assertTrue(stopped.await(10, TimeUnit.SECONDS));
Expand All @@ -335,6 +348,7 @@ public void testStartAsyncThenBlockingReadThenTimeout() throws Exception
CountDownLatch started = new CountDownLatch(1);
CountDownLatch completed = new CountDownLatch(1);
CountDownLatch stopped = new CountDownLatch(1);
AtomicReference<Thread> threadRef = new AtomicReference<>();
AtomicReference<Throwable> readException = new AtomicReference<>();
AbstractHandler handler = new AbstractHandler()
{
Expand All @@ -347,16 +361,15 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
AsyncContext async = request.startAsync();
async.setTimeout(100);

new Thread(() ->
Thread thread = new Thread(() ->
{
try
{
int b = baseRequest.getHttpInput().read();
if (b == '1')
{
started.countDown();
completed.await(10, TimeUnit.SECONDS);
Thread.sleep(500);
assertTrue(completed.await(10, TimeUnit.SECONDS));
if (baseRequest.getHttpInput().read() > Integer.MIN_VALUE)
throw new IllegalStateException();
}
Expand All @@ -366,14 +379,20 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
readException.set(t);
stopped.countDown();
}
}).start();
});
threadRef.set(thread);
thread.start();

try
{
// wait for thread to start and read first byte
started.await(10, TimeUnit.SECONDS);
assertTrue(started.await(10, TimeUnit.SECONDS));
// give it time to block on second byte
Thread.sleep(1000);
await().atMost(5, TimeUnit.SECONDS).until(() ->
{
Thread.State state = thread.getState();
return state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING;
});
}
catch (Throwable e)
{
Expand Down Expand Up @@ -406,7 +425,7 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
assertThat(response.getContent(), containsString("AsyncContext timeout"));

completed.countDown();
Thread.sleep(1000);
await().atMost(5, TimeUnit.SECONDS).until(() -> threadRef.get().getState() == Thread.State.TERMINATED);

// Async thread should have stopped
assertTrue(stopped.await(10, TimeUnit.SECONDS));
Expand All @@ -428,7 +447,7 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
baseRequest.setHandled(true);
if (baseRequest.getDispatcherType() != DispatcherType.ERROR)
{
new Thread(() ->
Thread thread = new Thread(() ->
{
try
{
Expand All @@ -445,14 +464,19 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
readException.set(t);
stopped.countDown();
}
}).start();
});
thread.start();

try
{
// wait for thread to start and read first byte
started.await(10, TimeUnit.SECONDS);
assertTrue(started.await(10, TimeUnit.SECONDS));
// give it time to block on second byte
Thread.sleep(1000);
await().atMost(5, TimeUnit.SECONDS).until(() ->
{
Thread.State state = thread.getState();
return state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING;
});
}
catch (Throwable e)
{
Expand Down Expand Up @@ -505,7 +529,7 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
baseRequest.setHandled(true);
response.setStatus(200);
response.setContentType("text/plain");
new Thread(() ->
Thread thread = new Thread(() ->
{
try
{
Expand All @@ -523,14 +547,19 @@ public void handle(String target, Request baseRequest, HttpServletRequest reques
readException.set(t);
stopped.countDown();
}
}).start();
});
thread.start();

try
{
// wait for thread to start and read first byte
started.await(10, TimeUnit.SECONDS);
assertTrue(started.await(10, TimeUnit.SECONDS));
// give it time to block on write
Thread.sleep(1000);
await().atMost(5, TimeUnit.SECONDS).until(() ->
{
Thread.State state = thread.getState();
return state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING;
});
}
catch (Throwable e)
{
Expand Down
Loading