Switch CRT HTTP clients (sync and async) from pull-based body to push-based writeData

.changes/next-release/bugfix-AWSCRTHTTPClient-aee08c2.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS CRT HTTP Client",
+    "contributor": "",
+    "description": "Fixed a potential deadlock in `AwsCrtHttpClient` that could occur when the request body `InputStream` blocked waiting for data on the CRT event loop thread. This could happen when a blocking stream (e.g., a `BufferedInputStream` wrapping a `ResponseInputStream`) was used as a request body and the read depended on the same event loop thread to deliver data. Request body writing now happens on the caller thread."
+}

http-clients/aws-crt-client/pom.xml

@@ -213,6 +213,32 @@
                     </archive>
                 </configuration>
             </plugin>
+            <!-- The Reactive Streams TCK tests are based on TestNG. See http://maven.apache.org/surefire/maven-surefire-plugin/examples/testng.html#Running_TestNG_and_JUnit_Tests -->
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <version>${maven.surefire.version}</version>
+                <configuration>
+                    <properties>
+                        <property>
+                            <name>junit</name>
+                            <value>false</value>
+                        </property>
+                    </properties>
+                </configuration>
+                <dependencies>
+                    <dependency>
+                        <groupId>org.apache.maven.surefire</groupId>
+                        <artifactId>surefire-junit-platform</artifactId>
+                        <version>${maven.surefire.version}</version>
+                    </dependency>
+                    <dependency>
+                        <groupId>org.apache.maven.surefire</groupId>
+                        <artifactId>surefire-testng</artifactId>
+                        <version>${maven.surefire.version}</version>
+                    </dependency>
+                </dependencies>
+            </plugin>
         </plugins>
     </build>
 </project>

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/AwsCrtHttpClient.java

@@ -18,13 +18,16 @@
 import static software.amazon.awssdk.http.HttpMetric.HTTP_CLIENT_NAME;
 
 import java.io.IOException;
+import java.io.InputStream;
 import java.time.Duration;
+import java.util.Arrays;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.function.Consumer;
 import software.amazon.awssdk.annotations.SdkPublicApi;
 import software.amazon.awssdk.crt.http.HttpException;
 import software.amazon.awssdk.crt.http.HttpStreamManager;
+import software.amazon.awssdk.http.ContentStreamProvider;
 import software.amazon.awssdk.http.ExecutableHttpRequest;
 import software.amazon.awssdk.http.HttpExecuteRequest;
 import software.amazon.awssdk.http.HttpExecuteResponse;
@@ -35,6 +38,8 @@
 import software.amazon.awssdk.http.crt.internal.AwsCrtClientBuilderBase;
 import software.amazon.awssdk.http.crt.internal.CrtRequestContext;
 import software.amazon.awssdk.http.crt.internal.CrtRequestExecutor;
+import software.amazon.awssdk.http.crt.internal.CrtStreamHandler;
+import software.amazon.awssdk.http.crt.internal.CrtUtils;
 import software.amazon.awssdk.utils.AttributeMap;
 import software.amazon.awssdk.utils.CompletableFutureUtils;
 
@@ -107,6 +112,8 @@ public ExecutableHttpRequest prepareRequest(HttpExecuteRequest request) {
     }
 
     private static final class CrtHttpRequest implements ExecutableHttpRequest {
+        private static final int WRITE_BUFFER_SIZE = 16 * 1024;
+
         private final CrtRequestContext context;
         private volatile CompletableFuture<SdkHttpFullResponse> responseFuture;
 
@@ -119,7 +126,10 @@ public HttpExecuteResponse call() throws IOException {
             HttpExecuteResponse.Builder builder = HttpExecuteResponse.builder();
 
             try {
-                responseFuture = new CrtRequestExecutor().execute(context);
+                CrtRequestExecutor.Result result = new CrtRequestExecutor().execute(context);
+                responseFuture = result.responseFuture();
+                writeRequestBody(result.streamHandler());
+
                 SdkHttpFullResponse response = CompletableFutureUtils.joinInterruptibly(responseFuture);
                 builder.response(response);
                 builder.responseBody(response.content().orElse(null));
@@ -140,6 +150,10 @@ public HttpExecuteResponse call() throws IOException {
                 }
 
                 if (cause instanceof HttpException) {
+                    Throwable wrapped = CrtUtils.wrapCrtException(cause);
+                    if (wrapped instanceof IOException) {
+                        throw (IOException) wrapped;
+                    }
                     throw (HttpException) cause;
                 }
 
@@ -151,6 +165,24 @@ public HttpExecuteResponse call() throws IOException {
             }
         }
 
+        private void writeRequestBody(CrtStreamHandler streamHandler) throws IOException {
+            ContentStreamProvider provider = context.sdkRequest().contentStreamProvider().orElse(null);
+            if (provider == null) {
+                return;
+            }
+
+            streamHandler.waitForStream();
+            try (InputStream inputStream = provider.newStream()) {
+                byte[] buf = new byte[WRITE_BUFFER_SIZE];
+                int read;
+                while ((read = inputStream.read(buf, 0, buf.length)) >= 0) {
+                    byte[] chunk = read == buf.length ? buf : Arrays.copyOf(buf, read);
+                    CompletableFutureUtils.joinInterruptibly(streamHandler.writeData(chunk, false));
+                }
+                CompletableFutureUtils.joinInterruptibly(streamHandler.writeData(null, true));
+            }
+        }
+
         @Override
         public void abort() {
             if (responseFuture != null) {

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/CrtAsyncRequestExecutor.java

@@ -26,33 +26,32 @@
 import software.amazon.awssdk.crt.http.HttpStreamBaseResponseHandler;
 import software.amazon.awssdk.http.SdkCancellationException;
 import software.amazon.awssdk.http.async.AsyncExecuteRequest;
-import software.amazon.awssdk.http.async.SdkAsyncHttpResponseHandler;
+import software.amazon.awssdk.http.crt.internal.request.CrtRequestBodyPublisherSubscriber;
 import software.amazon.awssdk.http.crt.internal.response.CrtResponseAdapter;
 import software.amazon.awssdk.metrics.MetricCollector;
 import software.amazon.awssdk.metrics.NoOpMetricCollector;
-import software.amazon.awssdk.utils.Logger;
 
 @SdkInternalApi
 public final class CrtAsyncRequestExecutor {
 
-    private static final Logger log = Logger.loggerFor(CrtAsyncRequestExecutor.class);
-
     public CompletableFuture<Void> execute(CrtAsyncRequestContext executionContext) {
         AsyncExecuteRequest asyncRequest = executionContext.sdkRequest();
         CompletableFuture<Void> requestFuture = createAsyncExecutionFuture(asyncRequest);
+        ResponseHandlerErrorNotifier errorNotifier = new ResponseHandlerErrorNotifier(asyncRequest.responseHandler());
 
         try {
-            doExecute(executionContext, asyncRequest, requestFuture);
+            doExecute(executionContext, asyncRequest, requestFuture, errorNotifier);
         } catch (Throwable t) {
-            reportAsyncFailure(t, requestFuture, asyncRequest.responseHandler());
+            reportAsyncFailure(t, requestFuture, errorNotifier);
         }
 
         return requestFuture;
     }
 
     private void doExecute(CrtAsyncRequestContext executionContext,
                            AsyncExecuteRequest asyncRequest,
-                           CompletableFuture<Void> requestFuture) {
+                           CompletableFuture<Void> requestFuture,
+                           ResponseHandlerErrorNotifier errorNotifier) {
         MetricCollector metricCollector = executionContext.metricCollector();
         boolean shouldPublishMetrics = metricCollector != null && !(metricCollector instanceof NoOpMetricCollector);
 
@@ -67,24 +66,48 @@ private void doExecute(CrtAsyncRequestContext executionContext,
 
         HttpRequestBase crtRequest = toAsyncCrtRequest(executionContext);
 
+        CompletableFuture<HttpStreamBase> streamFuture = new CompletableFuture<>();
+        CrtStreamHandler streamHandler = new CrtStreamHandler(streamFuture);
+
         HttpStreamBaseResponseHandler crtResponseHandler =
-            CrtResponseAdapter.toCrtResponseHandler(requestFuture, asyncRequest.responseHandler());
+            CrtResponseAdapter.toCrtResponseHandler(requestFuture, asyncRequest.responseHandler(), streamHandler, errorNotifier);
 
-        CompletableFuture<HttpStreamBase> streamFuture =
-            executionContext.streamManager().acquireStream(crtRequest, crtResponseHandler);
+        CrtRequestBodyPublisherSubscriber bodySubscriber =
+            new CrtRequestBodyPublisherSubscriber(streamHandler, requestFuture, errorNotifier);
 
         long finalAcquireStartTime = acquireStartTime;
 
-        streamFuture.whenComplete((stream, throwable) -> {
-            if (shouldPublishMetrics) {
-                reportMetrics(executionContext.streamManager(), metricCollector, finalAcquireStartTime);
-            }
+        executionContext.streamManager().acquireStream(crtRequest, crtResponseHandler, true)
+                        .handle((stream, throwable) -> {
+                            if (shouldPublishMetrics) {
+                                reportMetrics(executionContext.streamManager(), metricCollector, finalAcquireStartTime);
+                            }
+
+                            if (throwable != null) {
+                                handleAcquireFailure(throwable, streamFuture, requestFuture, errorNotifier);
+                                return null;
+                            }
+                            try {
+                                stream.activate();
+                                streamFuture.complete(stream);
+                                asyncRequest.requestContentPublisher().subscribe(bodySubscriber);
+                            } catch (Throwable t) {
+                                handleAcquireFailure(t, streamFuture, requestFuture, errorNotifier);
+                            }
+                            return null;
+                        }).exceptionally(t -> {
+                            handleAcquireFailure(t, streamFuture, requestFuture, errorNotifier);
+                            return null;
+                        });
+    }
 
-            if (throwable != null) {
-                Throwable toThrow = wrapCrtException(throwable);
-                reportAsyncFailure(toThrow, requestFuture, asyncRequest.responseHandler());
-            }
-        });
+    private void handleAcquireFailure(Throwable t,
+                                      CompletableFuture<HttpStreamBase> streamFuture,
+                                      CompletableFuture<Void> requestFuture,
+                                      ResponseHandlerErrorNotifier errorNotifier) {
+        Throwable toThrow = wrapCrtException(t);
+        streamFuture.completeExceptionally(toThrow);
+        reportAsyncFailure(toThrow, requestFuture, errorNotifier);
     }
 
     /**
@@ -109,18 +132,10 @@ private CompletableFuture<Void> createAsyncExecutionFuture(AsyncExecuteRequest r
         return future;
     }
 
-    /**
-     * Notify the provided response handler and future of the failure.
-     */
     private void reportAsyncFailure(Throwable cause,
                                     CompletableFuture<Void> executeFuture,
-                                    SdkAsyncHttpResponseHandler responseHandler) {
-        try {
-            responseHandler.onError(cause);
-        } catch (Exception e) {
-            log.error(() -> "SdkAsyncHttpResponseHandler " + responseHandler + " threw an exception in onError. It will be "
-                            + "ignored.", e);
-        }
+                                    ResponseHandlerErrorNotifier errorNotifier) {
+        errorNotifier.tryNotifyError(cause);
         executeFuture.completeExceptionally(cause);
     }
 }

http-clients/aws-crt-client/src/main/java/software/amazon/awssdk/http/crt/internal/CrtRequestExecutor.java

@@ -22,7 +22,6 @@
 import software.amazon.awssdk.annotations.SdkInternalApi;
 import software.amazon.awssdk.crt.http.HttpRequest;
 import software.amazon.awssdk.crt.http.HttpStreamBase;
-import software.amazon.awssdk.crt.http.HttpStreamBaseResponseHandler;
 import software.amazon.awssdk.http.SdkHttpFullResponse;
 import software.amazon.awssdk.http.crt.internal.request.CrtRequestAdapter;
 import software.amazon.awssdk.http.crt.internal.response.InputStreamAdaptingHttpStreamResponseHandler;
@@ -32,49 +31,91 @@
 @SdkInternalApi
 public final class CrtRequestExecutor {
 
-    public CompletableFuture<SdkHttpFullResponse> execute(CrtRequestContext executionContext) {
-        CompletableFuture<SdkHttpFullResponse> requestFuture = new CompletableFuture<>();
+    public Result execute(CrtRequestContext executionContext) {
+        CompletableFuture<SdkHttpFullResponse> responseFuture = new CompletableFuture<>();
+        CompletableFuture<HttpStreamBase> streamFuture = new CompletableFuture<>();
+        CrtStreamHandler streamHandler = new CrtStreamHandler(streamFuture);
 
         try {
-            doExecute(executionContext, requestFuture);
+            doExecute(executionContext, responseFuture, streamHandler, streamFuture);
         } catch (Throwable t) {
-            requestFuture.completeExceptionally(t);
+            // Fail streamFuture too so any caller blocked in waitForStream() unblocks.
+            streamFuture.completeExceptionally(t);
+            responseFuture.completeExceptionally(t);
         }
 
-        return requestFuture;
+        return new Result(responseFuture, streamHandler);
     }
 
-    private void doExecute(CrtRequestContext executionContext, CompletableFuture<SdkHttpFullResponse> requestFuture) {
+    private void doExecute(CrtRequestContext executionContext,
+                           CompletableFuture<SdkHttpFullResponse> responseFuture,
+                           CrtStreamHandler streamHandler,
+                           CompletableFuture<HttpStreamBase> streamFuture) {
         MetricCollector metricCollector = executionContext.metricCollector();
         boolean shouldPublishMetrics = metricCollector != null && !(metricCollector instanceof NoOpMetricCollector);
 
         long acquireStartTime = 0;
 
         if (shouldPublishMetrics) {
-            // go ahead and get acquireStartTime for the concurrency timer as early as possible,
-            // so it's as accurate as possible, but only do it in a branch since clock_gettime()
-            // results in a full sys call barrier (multiple mutexes and a hw interrupt).
             acquireStartTime = System.nanoTime();
         }
 
-        HttpStreamBaseResponseHandler crtResponseHandler = new InputStreamAdaptingHttpStreamResponseHandler(requestFuture);
+        InputStreamAdaptingHttpStreamResponseHandler crtResponseHandler =
+            new InputStreamAdaptingHttpStreamResponseHandler(responseFuture, streamHandler);
 
         HttpRequest crtRequest = CrtRequestAdapter.toCrtRequest(executionContext);
 
-        CompletableFuture<HttpStreamBase> streamFuture =
-            executionContext.streamManager().acquireStream(crtRequest, crtResponseHandler);
+        boolean hasBody = executionContext.sdkRequest().contentStreamProvider().isPresent();
 
         long finalAcquireStartTime = acquireStartTime;
 
-        streamFuture.whenComplete((streamBase, throwable) -> {
-            if (shouldPublishMetrics) {
-                reportMetrics(executionContext.streamManager(), metricCollector, finalAcquireStartTime);
-            }
+        executionContext.streamManager().acquireStream(crtRequest, crtResponseHandler, hasBody)
+                        .handle((streamBase, throwable) -> {
+                            if (shouldPublishMetrics) {
+                                reportMetrics(executionContext.streamManager(), metricCollector, finalAcquireStartTime);
+                            }
+
+                            if (throwable != null) {
+                                handleAcquireFailure(throwable, streamFuture, responseFuture);
+                                return null;
+                            }
+                            try {
+                                streamBase.activate();
+                                streamFuture.complete(streamBase);
+                            } catch (Throwable t) {
+                                handleAcquireFailure(t, streamFuture, responseFuture);
+                            }
+                            return null;
+                        }).exceptionally(t -> {
+                            handleAcquireFailure(t, streamFuture, responseFuture);
+                            return null;
+                        });
+    }
+
+    private void handleAcquireFailure(Throwable t,
+                                      CompletableFuture<HttpStreamBase> streamFuture,
+                                      CompletableFuture<SdkHttpFullResponse> responseFuture) {
+        Throwable toThrow = wrapCrtException(t);
+        streamFuture.completeExceptionally(toThrow);
+        responseFuture.completeExceptionally(toThrow);
+    }
 
-            if (throwable != null) {
-                Throwable toThrow = wrapCrtException(throwable);
-                requestFuture.completeExceptionally(toThrow);
-            }
-        });
+    @SdkInternalApi
+    public static final class Result {
+        private final CompletableFuture<SdkHttpFullResponse> responseFuture;
+        private final CrtStreamHandler streamHandler;
+
+        private Result(CompletableFuture<SdkHttpFullResponse> responseFuture, CrtStreamHandler streamHandler) {
+            this.responseFuture = responseFuture;
+            this.streamHandler = streamHandler;
+        }
+
+        public CompletableFuture<SdkHttpFullResponse> responseFuture() {
+            return responseFuture;
+        }
+
+        public CrtStreamHandler streamHandler() {
+            return streamHandler;
+        }
     }
 }