7449 out of order chunk 2x (helidon-io#7460)

* Fixed problem in AUTO_FLUSH backpressure strategy (helidon-io#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 <[email protected]> * Use constant. Signed-off-by: Santiago Pericasgeertsen <[email protected]> --------- Signed-off-by: Santiago Pericasgeertsen <[email protected]> * Fix intermittent out-of-order chunk helidon-io#7407 (helidon-io#7441) Signed-off-by: Daniel Kec <[email protected]> Co-authored-by: Santiago Pericas-Geertsen <[email protected]> --------- Signed-off-by: Santiago Pericasgeertsen <[email protected]> Signed-off-by: Daniel Kec <[email protected]> Co-authored-by: Santiago Pericas-Geertsen <[email protected]>
tomas-langer · Aug 25, 2023 · 832d35e · 832d35e
1 parent b89cf88
commit 832d35e
Show file tree

Hide file tree

Showing 4 changed files with 120 additions and 12 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
@@ -0,0 +1,102 @@
+/*
+ * 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 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 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;
+
+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 byte[] sha256;
+    private static final int FILE_SIZE = 40 * 1024 * 1024;
+    private static final int BUFFER_SIZE = 8 * 1024;
+
+    @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, NoSuchAlgorithmException {
+        file = File.createTempFile("file", ".bin");
+        file.deleteOnExit();
+        MessageDigest messageDigest = MessageDigest.getInstance("SHA-256");
+        try (FileOutputStream fos = new FileOutputStream(file)) {
+            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() 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)));
+
+            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
@@ -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/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);
         }
     }
 

diff --git 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,15 +135,15 @@ void linear() {
         }
     }
 
-    @Test
+    @Test(invocationCount = 10)
     void autoFlush() {
         AtomicLong receivedSize = new AtomicLong(0);
 
         WebServer webServer = null;
         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<String> 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));