error pending futures out on cancel

Author: DmitriyMusatkin

include/aws/s3/private/s3_meta_request_impl.h

@@ -227,6 +227,9 @@ struct aws_s3_meta_request {
         /* To track aws_s3_requests with cancellable HTTP streams */
         struct aws_linked_list cancellable_http_streams_list;
 
+        /* To track aws_future_s3_buffers that might need to be cleaned up on cancel */
+        struct aws_linked_list pending_buffer_futures;
+
         /* Data for async-writes. */
         struct aws_s3_async_write_data {
             /* Whether a part request can be sent (we have 1 part's worth of data, or EOF) */
@@ -403,6 +406,9 @@ bool aws_s3_meta_request_are_events_out_for_delivery_synced(struct aws_s3_meta_r
 /* Cancel the requests with cancellable HTTP stream for the meta request */
 void aws_s3_meta_request_cancel_cancellable_requests_synced(struct aws_s3_meta_request *meta_request, int error_code);
 
+/* Cancel the pending buffer futures for the meta request */
+void aws_s3_meta_request_cancel_pending_buffer_futures_synced(struct aws_s3_meta_request *meta_request, int error_code);
+
 /* Asynchronously read from the meta request's input stream. Should always be done outside of any mutex,
  * as reading from the stream could cause user code to call back into aws-c-s3.
  * This will fill the buffer to capacity, unless end of stream is reached.

include/aws/s3/private/s3_request.h

@@ -114,10 +114,16 @@ struct aws_s3_request {
     /* Linked list node used for tracking the request is active from HTTP level. */
     struct aws_linked_list_node cancellable_http_streams_list_node;
 
+    /* Linked list node used for tracking buffer acquire futures. */
+    struct aws_linked_list_node pending_buffer_future_list_node;
+
     /* The meta request lock must be held to access the data */
     struct {
         /* The underlying http stream, only valid when the request is active from HTTP level */
         struct aws_http_stream *cancellable_http_stream;
+
+        /* Buffer future. */
+        struct aws_future_s3_buffer_ticket *buffer_future;
     } synced_data;
 
     /* TODO Ref count on the request is no longer needed--only one part of code should ever be holding onto a request,

source/s3_auto_ranged_put.c

@@ -1743,6 +1743,7 @@ static int s_s3_auto_ranged_put_pause(
     aws_s3_meta_request_set_fail_synced(meta_request, NULL, AWS_ERROR_S3_PAUSED);
 
     aws_s3_meta_request_cancel_cancellable_requests_synced(meta_request, AWS_ERROR_S3_PAUSED);
+    aws_s3_meta_request_cancel_pending_buffer_futures_synced(meta_request, AWS_ERROR_S3_PAUSED);
 
     /* unlock */
     aws_s3_meta_request_unlock_synced_data(meta_request);

source/s3_client.c

@@ -1860,16 +1860,26 @@ static void s_on_pool_buffer_reserved(void *user_data) {
 
     request->ticket = aws_future_s3_buffer_ticket_get_result_by_move(future_ticket);
 
+    /* BEGIN CRITICAL SECTION */
+    {
+        aws_s3_meta_request_lock_synced_data(meta_request);
+        AWS_FATAL_ASSERT(request->synced_data.buffer_future);
+        aws_linked_list_remove(&request->pending_buffer_future_list_node);
+        request->synced_data.buffer_future = aws_future_s3_buffer_ticket_release(request->synced_data.buffer_future);
+        aws_s3_meta_request_unlock_synced_data(meta_request);
+    }
+    /* END CRITICAL SECTION */
+
     aws_future_s3_buffer_ticket_release(payload->buffer_future);
     aws_mem_release(payload->allocator, payload);
 
-    aws_s3_meta_request_prepare_request(request->meta_request, 
-        request, payload->callback, payload->user_data);
+    aws_s3_meta_request_prepare_request(request->meta_request, request, payload->callback, payload->user_data);
     return;
 }
 
-void s_acquire_mem_and_prepare_request(struct aws_s3_client *client, 
-    struct aws_s3_request *request, 
+void s_acquire_mem_and_prepare_request(
+    struct aws_s3_client *client,
+    struct aws_s3_request *request,
     aws_s3_meta_request_prepare_request_callback_fn *callback,
     void *user_data) {
 
@@ -1879,14 +1889,25 @@ void s_acquire_mem_and_prepare_request(struct aws_s3_client *client,
         struct aws_s3_buffer_pool_reserve_meta meta = {
             .client = client, .meta_request = meta_request, .size = meta_request->part_size};
 
-        struct aws_s3_reserve_memory_payload *payload = aws_mem_calloc(allocator, 1, sizeof(struct aws_s3_reserve_memory_payload));
+        struct aws_s3_reserve_memory_payload *payload =
+            aws_mem_calloc(allocator, 1, sizeof(struct aws_s3_reserve_memory_payload));
 
         payload->allocator = allocator;
         payload->request = request;
         payload->callback = callback;
         payload->user_data = user_data;
         payload->buffer_future = aws_s3_buffer_pool_reserve(request->meta_request->client->buffer_pool, meta);
 
+        /* BEGIN CRITICAL SECTION */
+        {
+            aws_s3_meta_request_lock_synced_data(meta_request);
+            aws_linked_list_push_back(
+                &meta_request->synced_data.pending_buffer_futures, &request->pending_buffer_future_list_node);
+            request->synced_data.buffer_future = aws_future_s3_buffer_ticket_acquire(payload->buffer_future);
+            aws_s3_meta_request_unlock_synced_data(meta_request);
+        }
+        /* END CRITICAL SECTION */
+
         aws_future_s3_buffer_ticket_register_callback(payload->buffer_future, s_on_pool_buffer_reserved, payload);
         return;
     }

source/s3_meta_request.c

@@ -180,6 +180,7 @@ int aws_s3_meta_request_init_base(
     /* Set up reference count. */
     aws_ref_count_init(&meta_request->ref_count, meta_request, s_s3_meta_request_destroy);
     aws_linked_list_init(&meta_request->synced_data.cancellable_http_streams_list);
+    aws_linked_list_init(&meta_request->synced_data.pending_buffer_futures);
 
     if (part_size == SIZE_MAX) {
         aws_raise_error(AWS_ERROR_INVALID_ARGUMENT);
@@ -377,6 +378,7 @@ void aws_s3_meta_request_cancel(struct aws_s3_meta_request *meta_request) {
     aws_s3_meta_request_lock_synced_data(meta_request);
     aws_s3_meta_request_set_fail_synced(meta_request, NULL, AWS_ERROR_S3_CANCELED);
     aws_s3_meta_request_cancel_cancellable_requests_synced(meta_request, AWS_ERROR_S3_CANCELED);
+    aws_s3_meta_request_cancel_pending_buffer_futures_synced(meta_request, AWS_ERROR_S3_CANCELED);
     aws_s3_meta_request_unlock_synced_data(meta_request);
     /* END CRITICAL SECTION */
 
@@ -537,6 +539,7 @@ static void s_s3_meta_request_destroy(void *user_data) {
     aws_array_list_clean_up(&meta_request->io_threaded_data.event_delivery_array);
 
     AWS_ASSERT(aws_linked_list_empty(&meta_request->synced_data.cancellable_http_streams_list));
+    AWS_ASSERT(aws_linked_list_empty(&meta_request->synced_data.pending_buffer_futures));
 
     aws_s3_meta_request_result_clean_up(meta_request, &meta_request->synced_data.finish_result);
 
@@ -1825,6 +1828,22 @@ void aws_s3_meta_request_cancel_cancellable_requests_synced(struct aws_s3_meta_r
     }
 }
 
+void aws_s3_meta_request_cancel_pending_buffer_futures_synced(
+    struct aws_s3_meta_request *meta_request,
+    int error_code) {
+    ASSERT_SYNCED_DATA_LOCK_HELD(meta_request);
+    while (!aws_linked_list_empty(&meta_request->synced_data.pending_buffer_futures)) {
+
+        struct aws_linked_list_node *request_node =
+            aws_linked_list_pop_front(&meta_request->synced_data.pending_buffer_futures);
+
+        struct aws_s3_request *request =
+            AWS_CONTAINER_OF(request_node, struct aws_s3_request, pending_buffer_future_list_node);
+
+        aws_future_s3_buffer_ticket_set_error(request->synced_data.buffer_future, error_code);
+    }
+}
+
 static struct aws_s3_request_metrics *s_s3_request_finish_up_and_release_metrics(
     struct aws_s3_request_metrics *metrics,
     struct aws_s3_meta_request *meta_request) {