From ebc41fe91856f994ebc54778089a790cba390a33 Mon Sep 17 00:00:00 2001 From: Daniel Kec Date: Thu, 24 Aug 2023 14:20:55 +0200 Subject: [PATCH] Fix intermittent out-of-order chunk #7407 --- .../microprofile/server/AutoFlushTest.java | 48 +++++++++++++++---- .../io/helidon/webserver/NettyChannel.java | 13 ++--- .../webserver/WaterMarkedBackpressureIT.java | 13 ++++- 3 files changed, 54 insertions(+), 20 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..bdcb65058ba 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,6 +21,9 @@ 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; @@ -32,7 +35,7 @@ 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 org.junit.jupiter.api.RepeatedTest; import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; @@ -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,48 @@ 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", new File("/tmp")); 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); + for (int i = 0; i < FILE_SIZE / BUFFER_SIZE; i++) { + byte[] bytes = dataBuffer(BUFFER_SIZE, (byte) (i % 10 + '0')); + fos.write(bytes); + messageDigest.update(bytes); } } + sha256 = messageDigest.digest(); } - @Test - void testAutoFlush() { + static byte[] dataBuffer(int size, byte b) { + byte[] bytes = new byte[size]; + Arrays.fill(bytes, b); + bytes[size - 1] = '\n'; // to open file in editor + return bytes; + } + + @RepeatedTest(10) + void testAutoFlush() throws NoSuchAlgorithmException, IOException { 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); + File downloaded = File.createTempFile("downloaded", ".bin", new File("/tmp")); + try (FileOutputStream fos = new FileOutputStream(downloaded)) { + fos.write(file); + } + 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 5e6cff39bac..80801808cb3 100644 --- a/webserver/webserver/src/main/java/io/helidon/webserver/NettyChannel.java +++ b/webserver/webserver/src/main/java/io/helidon/webserver/NettyChannel.java @@ -75,14 +75,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 adc2adaaab5..6fc97f710fc 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(r -> r.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));