From 4da94b949792ed481ce42cce19720e25cc4340de Mon Sep 17 00:00:00 2001 From: Santiago Pericas-Geertsen Date: Wed, 5 Apr 2023 12:39:55 -0400 Subject: [PATCH 1/2] Fixed problem in AUTO_FLUSH backpressure strategy (#6556) * Fixed problem in AUTO_FLUSH strategy that may result in pub-sub deadlock. Increment buffer sum before checking watermark and flushing. * Generate large binary file programmatically. Signed-off-by: Santiago Pericasgeertsen * Use constant. Signed-off-by: Santiago Pericasgeertsen --------- Signed-off-by: Santiago Pericasgeertsen --- .../microprofile/server/AutoFlushTest.java | 85 +++++++++++++++++++ .../webserver/ServerResponseSubscription.java | 4 +- 2 files changed, 87 insertions(+), 2 deletions(-) create mode 100644 microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java diff --git a/microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java b/microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java new file mode 100644 index 00000000000..0650c130499 --- /dev/null +++ b/microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java @@ -0,0 +1,85 @@ +/* + * Copyright (c) 2023 Oracle and/or its affiliates. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.helidon.microprofile.server; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.InputStream; + +import io.helidon.common.http.Http; +import io.helidon.microprofile.tests.junit5.AddBean; +import io.helidon.microprofile.tests.junit5.HelidonTest; +import jakarta.inject.Inject; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.client.WebTarget; +import jakarta.ws.rs.core.Response; +import jakarta.ws.rs.core.StreamingOutput; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +import static org.hamcrest.CoreMatchers.is; +import static org.hamcrest.MatcherAssert.assertThat; + +@HelidonTest +@AddBean(AutoFlushTest.AutoFlushResource.class) +class AutoFlushTest { + + private static File file; + private static final int FILE_SIZE = 6000000; + + @Inject + private WebTarget target; + + @Path("auto-flush") + public static class AutoFlushResource { + + @GET + public Response getDefaultMessage() { + return Response.ok((StreamingOutput) outputStream -> { + try (InputStream is = new FileInputStream(file)) { + is.transferTo(outputStream); + } + }).header(Http.Header.CONTENT_LENGTH, String.valueOf(FILE_SIZE)).build(); + } + } + + @BeforeAll + static void createTempFile() throws IOException { + file = File.createTempFile("zero", ".bin"); + file.deleteOnExit(); + byte[] zero = new byte[10 * 1000]; + try (FileOutputStream fos = new FileOutputStream(file)) { + for (int i = 0; i < FILE_SIZE / zero.length; i++) { + fos.write(zero); + } + } + } + + @Test + void testAutoFlush() { + try (Response resp = target.path("auto-flush") + .request() + .get()) { + assertThat(resp.getStatus(), is(200)); + assertThat(resp.getHeaderString(Http.Header.CONTENT_LENGTH), is(String.valueOf(FILE_SIZE))); + assertThat(resp.readEntity(byte[].class).length, is(FILE_SIZE)); + } + } +} diff --git a/webserver/webserver/src/main/java/io/helidon/webserver/ServerResponseSubscription.java b/webserver/webserver/src/main/java/io/helidon/webserver/ServerResponseSubscription.java index 0c85b05c8b7..b7195c4328c 100644 --- a/webserver/webserver/src/main/java/io/helidon/webserver/ServerResponseSubscription.java +++ b/webserver/webserver/src/main/java/io/helidon/webserver/ServerResponseSubscription.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2022 Oracle and/or its affiliates. + * Copyright (c) 2022, 2023 Oracle and/or its affiliates. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -120,10 +120,10 @@ class WatermarkAutoFlush extends WatermarkLinear { @Override public void inc(NettyChannel channel, int byteSize) { + super.inc(channel, byteSize); if (!watermarkNotReached()) { channel.flush(); } - super.inc(channel, byteSize); } } From 5da162bdbac5a1143c0349a3d05f42d223bed468 Mon Sep 17 00:00:00 2001 From: Daniel Kec Date: Thu, 24 Aug 2023 21:43:35 +0200 Subject: [PATCH 2/2] Fix intermittent out-of-order chunk #7407 (#7441) Signed-off-by: Daniel Kec Co-authored-by: Santiago Pericas-Geertsen --- .../microprofile/server/AutoFlushTest.java | 45 +++++++++++++------ .../io/helidon/webserver/NettyChannel.java | 13 +++--- .../webserver/WaterMarkedBackpressureIT.java | 13 +++++- 3 files changed, 47 insertions(+), 24 deletions(-) diff --git a/microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java b/microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java index 0650c130499..dd7c838c9fc 100644 --- a/microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java +++ b/microprofile/server/src/test/java/io/helidon/microprofile/server/AutoFlushTest.java @@ -21,16 +21,19 @@ import java.io.FileOutputStream; import java.io.IOException; import java.io.InputStream; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.Arrays; import io.helidon.common.http.Http; import io.helidon.microprofile.tests.junit5.AddBean; import io.helidon.microprofile.tests.junit5.HelidonTest; -import jakarta.inject.Inject; -import jakarta.ws.rs.GET; -import jakarta.ws.rs.Path; -import jakarta.ws.rs.client.WebTarget; -import jakarta.ws.rs.core.Response; -import jakarta.ws.rs.core.StreamingOutput; +import javax.inject.Inject; +import javax.ws.rs.GET; +import javax.ws.rs.Path; +import javax.ws.rs.client.WebTarget; +import javax.ws.rs.core.Response; +import javax.ws.rs.core.StreamingOutput; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; @@ -42,7 +45,9 @@ class AutoFlushTest { private static File file; - private static final int FILE_SIZE = 6000000; + private static byte[] sha256; + private static final int FILE_SIZE = 40 * 1024 * 1024; + private static final int BUFFER_SIZE = 8 * 1024; @Inject private WebTarget target; @@ -61,25 +66,37 @@ public Response getDefaultMessage() { } @BeforeAll - static void createTempFile() throws IOException { - file = File.createTempFile("zero", ".bin"); + static void createTempFile() throws IOException, NoSuchAlgorithmException { + file = File.createTempFile("file", ".bin"); file.deleteOnExit(); - byte[] zero = new byte[10 * 1000]; + MessageDigest messageDigest = MessageDigest.getInstance("SHA-256"); try (FileOutputStream fos = new FileOutputStream(file)) { - for (int i = 0; i < FILE_SIZE / zero.length; i++) { - fos.write(zero); + byte[] bytes = new byte[BUFFER_SIZE]; + for (int i = 0; i < FILE_SIZE / BUFFER_SIZE; i++) { + Arrays.fill(bytes, (byte) (i % 10 + '0')); + fos.write(bytes); + messageDigest.update(bytes); } } + sha256 = messageDigest.digest(); } @Test - void testAutoFlush() { + void testAutoFlush() throws NoSuchAlgorithmException { try (Response resp = target.path("auto-flush") .request() .get()) { assertThat(resp.getStatus(), is(200)); assertThat(resp.getHeaderString(Http.Header.CONTENT_LENGTH), is(String.valueOf(FILE_SIZE))); - assertThat(resp.readEntity(byte[].class).length, is(FILE_SIZE)); + + byte[] file = resp.readEntity(byte[].class); + assertThat(file.length, is(FILE_SIZE)); + MessageDigest messageDigest = MessageDigest.getInstance("SHA-256"); + for (int i = 0; i < FILE_SIZE / BUFFER_SIZE; i++) { + messageDigest.update(file, i * BUFFER_SIZE, BUFFER_SIZE); + } + byte[] newSha256 = messageDigest.digest(); + assertThat(sha256, is(newSha256)); } } } diff --git a/webserver/webserver/src/main/java/io/helidon/webserver/NettyChannel.java b/webserver/webserver/src/main/java/io/helidon/webserver/NettyChannel.java index f5cd1f4b176..7e22bdb6796 100644 --- a/webserver/webserver/src/main/java/io/helidon/webserver/NettyChannel.java +++ b/webserver/webserver/src/main/java/io/helidon/webserver/NettyChannel.java @@ -76,14 +76,11 @@ void read() { * Request to flush all pending messages via this ChannelOutboundInvoker from Netty's event loop thread. */ void flush() { - writeFuture = writeFuture.thenApply(f -> { - if (channel.eventLoop().inEventLoop()) { - channel.flush(); - } else { - channel.eventLoop().execute(channel::flush); - } - return f; - }); + if (channel.eventLoop().inEventLoop()) { + channel.flush(); + } else { + channel.eventLoop().execute(channel::flush); + } } /** diff --git a/webserver/webserver/src/test/java/io/helidon/webserver/WaterMarkedBackpressureIT.java b/webserver/webserver/src/test/java/io/helidon/webserver/WaterMarkedBackpressureIT.java index ad4104b717e..5e7a6c2dae8 100644 --- a/webserver/webserver/src/test/java/io/helidon/webserver/WaterMarkedBackpressureIT.java +++ b/webserver/webserver/src/test/java/io/helidon/webserver/WaterMarkedBackpressureIT.java @@ -18,6 +18,9 @@ import java.nio.ByteBuffer; import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletionException; import java.util.concurrent.ExecutorService; @@ -39,6 +42,7 @@ import static io.helidon.webserver.BackpressureStrategy.AUTO_FLUSH; import static io.helidon.webserver.BackpressureStrategy.LINEAR; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.contains; import static org.hamcrest.Matchers.greaterThan; import static org.hamcrest.Matchers.is; @@ -131,7 +135,7 @@ void linear() { } } - @Test + @Test(invocationCount = 10) void autoFlush() { AtomicLong receivedSize = new AtomicLong(0); @@ -139,7 +143,7 @@ void autoFlush() { try { webServer = WebServer.builder() .host("localhost") - .backpressureBufferSize(500) + .backpressureBufferSize(200) .backpressureStrategy(AUTO_FLUSH) .routing(Routing.builder().get("/", (req, res) -> { res.send(Multi.range(0, 150) @@ -157,6 +161,8 @@ void autoFlush() { .start() .await(TIMEOUT); + List receivedData = new ArrayList<>(100); + WebClient.builder() .baseUri("http://localhost:" + webServer.port()) .build() @@ -169,6 +175,7 @@ void autoFlush() { receivedSize.addAndGet(bytes.length); String data = new String(bytes); chunk.release(); + receivedData.add(data); return !data.equals("00100"); }) .ignoreElements() @@ -178,6 +185,8 @@ void autoFlush() { }) .await(TIMEOUT); + assertThat(receivedData, contains(Multi.range(0, 101) + .map(l -> String.format("%05d", l)).collectList().await().toArray(String[]::new))); // Stochastic test as watermarking depends on Netty's flush callbacks assertThat(receivedSize.get(), greaterThan(499L));