Refactoring meta request types (#79)

* aws_s3_request is now in its own header/source file.
* All meta request types now get their own private header file to make writing tests for those easier. 
* Meta request types no longer use state machines for tracking what request is to be sent next.  Instead, a series of conditions is evaluated each time next_request is called.
* The final meta request result is now set with a success or fail function.  Once all work that the meta request currently has in progress completes, then the meta request internally calls aws_s3_meta_request_finish, which invokes the finish callback.
* Performance Behavior Improvement: added a mitigation for global request limit being hit.  Client will first ask for new requests from meta requests with a "conservative" flag, which will cause auto-ranged-gets to limit the amount of requests returned, preventing parts from getting "backed up" waiting for earlier parts.
* Bug fix: no header/body callbacks should now ever be invoked after the finish callback.
* The client handles when a meta request is scheduled for work, instead of the meta request having to manage that at all.

Author: rccarper

include/aws/s3/private/s3_auto_ranged_get.h

@@ -0,0 +1,39 @@
+#ifndef AWS_S3_AUTO_RANGED_GET_H
+#define AWS_S3_AUTO_RANGED_GET_H
+
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include "aws/s3/private/s3_meta_request_impl.h"
+
+enum aws_s3_auto_ranged_get_request_type {
+    AWS_S3_AUTO_RANGE_GET_REQUEST_TYPE_PART,
+};
+
+struct aws_s3_auto_ranged_get {
+    struct aws_s3_meta_request base;
+
+    /* Members to only be used when the mutex in the base type is locked. */
+    struct {
+        uint32_t total_num_parts;
+
+        uint32_t num_parts_requested;
+        uint32_t num_parts_completed;
+        uint32_t num_parts_successful;
+        uint32_t num_parts_failed;
+
+        size_t total_object_size;
+
+    } synced_data;
+};
+
+/* Creates a new auto-ranged get meta request.  This will do multiple parallel ranged-gets when appropriate. */
+struct aws_s3_meta_request *aws_s3_meta_request_auto_ranged_get_new(
+    struct aws_allocator *allocator,
+    struct aws_s3_client *client,
+    size_t part_size,
+    const struct aws_s3_meta_request_options *options);
+
+#endif

include/aws/s3/private/s3_auto_ranged_put.h

@@ -6,19 +6,7 @@
  * SPDX-License-Identifier: Apache-2.0.
  */
 
-#include "aws/s3/private/s3_client_impl.h"
 #include "aws/s3/private/s3_meta_request_impl.h"
-#include "aws/s3/private/s3_util.h"
-
-enum aws_s3_auto_ranged_put_state {
-    AWS_S3_AUTO_RANGED_PUT_STATE_START,
-    AWS_S3_AUTO_RANGED_PUT_STATE_WAITING_FOR_CREATE,
-    AWS_S3_AUTO_RANGED_PUT_STATE_SENDING_PARTS,
-    AWS_S3_AUTO_RANGED_PUT_STATE_WAITING_FOR_PARTS,
-    AWS_S3_AUTO_RANGED_PUT_STATE_WAITING_FOR_CANCEL,
-    AWS_S3_AUTO_RANGED_PUT_STATE_SEND_COMPLETE,
-    AWS_S3_AUTO_RANGED_PUT_STATE_WAITING_FOR_COMPLETE,
-};
 
 enum aws_s3_auto_ranged_put_request_tag {
     AWS_S3_AUTO_RANGED_PUT_REQUEST_TAG_CREATE_MULTIPART_UPLOAD,
@@ -35,25 +23,44 @@ struct aws_s3_auto_ranged_put {
 
     uint64_t content_length;
 
+    /* Only meant for use in the update function, which is never called concurrently. */
     struct {
-        /* Only meant for use in the next_request function, which is never called concurrently. */
         uint32_t next_part_number;
-    } threaded_next_request_data;
+    } threaded_update_data;
 
+    /* Members to only be used when the mutex in the base type is locked. */
     struct {
-        enum aws_s3_auto_ranged_put_state state;
         struct aws_array_list etag_list;
 
         uint32_t total_num_parts;
         uint32_t num_parts_sent;
         uint32_t num_parts_completed;
+        uint32_t num_parts_successful;
+        uint32_t num_parts_failed;
 
         struct aws_http_headers *needed_response_headers;
-        struct aws_s3_meta_request_finish_options *cached_finish_options;
 
-        bool create_multipart_upload_successful;
+        int create_multipart_upload_error_code;
+        int complete_multipart_upload_error_code;
+        int abort_multipart_upload_error_code;
+
+        uint32_t create_multipart_upload_sent : 1;
+        uint32_t create_multipart_upload_completed : 1;
+        uint32_t complete_multipart_upload_sent : 1;
+        uint32_t complete_multipart_upload_completed : 1;
+        uint32_t abort_multipart_upload_sent : 1;
+        uint32_t abort_multipart_upload_completed : 1;
 
     } synced_data;
 };
 
+/* Creates a new auto-ranged put meta request.  This will do a multipart upload in parallel when appropriate. */
+struct aws_s3_meta_request *aws_s3_meta_request_auto_ranged_put_new(
+    struct aws_allocator *allocator,
+    struct aws_s3_client *client,
+    size_t part_size,
+    uint64_t content_length,
+    uint32_t num_parts,
+    const struct aws_s3_meta_request_options *options);
+
 #endif

include/aws/s3/private/s3_client_impl.h

@@ -92,18 +92,16 @@ struct aws_s3_vip_connection {
     struct aws_retry_token *retry_token;
 
     /* True if the connection is currently retrying to process the request. */
-    bool is_retry;
+    uint32_t is_retry : 1;
+
+    uint32_t is_active : 1;
 };
 
 struct aws_s3_client_vtable {
 
     struct aws_s3_meta_request *(
         *meta_request_factory)(struct aws_s3_client *client, const struct aws_s3_meta_request_options *options);
 
-    void (*push_meta_request)(struct aws_s3_client *client, struct aws_s3_meta_request *meta_request);
-
-    void (*remove_meta_request)(struct aws_s3_client *client, struct aws_s3_meta_request *meta_request);
-
     void (*acquire_http_connection)(
         struct aws_s3_client *client,
         struct aws_s3_vip_connection *vip_connection,
@@ -168,6 +166,20 @@ struct aws_s3_client {
     aws_s3_client_shutdown_complete_callback_fn *shutdown_callback;
     void *shutdown_callback_user_data;
 
+    struct {
+        /* Number of requests being sent/received over network. */
+        struct aws_atomic_var num_requests_network_io;
+
+        /* Number of requests sitting in their meta request priority queue, waiting to be streamed. */
+        struct aws_atomic_var num_requests_queued_waiting;
+
+        /* Number of requests currently scheduled to be streamed or are actively being streamed. */
+        struct aws_atomic_var num_requests_streaming;
+
+        /* Number of allocated VIP connnections. */
+        struct aws_atomic_var num_allocated_vip_connections;
+    } stats;
+
     struct {
         struct aws_mutex lock;
 
@@ -230,19 +242,15 @@ struct aws_s3_client {
         /* Client list of on going meta requests. */
         struct aws_linked_list meta_requests;
 
-        /* Next meta request that the work_task will start with on its next update. */
-        struct aws_s3_meta_request *next_meta_request;
-
         /* Number of requests being processed, either still being sent/received or being streamed to the caller. */
         uint32_t num_requests_in_flight;
 
+        /* Number of current active aws_s3_vip_connections. */
+        uint32_t num_active_vip_connections;
+
     } threaded_data;
 };
 
-void aws_s3_client_push_meta_request(struct aws_s3_client *client, struct aws_s3_meta_request *meta_request);
-
-void aws_s3_client_remove_meta_request(struct aws_s3_client *client, struct aws_s3_meta_request *meta_request);
-
 int aws_s3_client_make_request(struct aws_s3_client *client, struct aws_s3_vip_connection *vip_connection);
 
 void aws_s3_client_notify_connection_finished(
@@ -253,10 +261,18 @@ void aws_s3_client_notify_connection_finished(
 
 void aws_s3_client_notify_request_destroyed(struct aws_s3_client *client, struct aws_s3_request *request);
 
+typedef void(aws_s3_client_stream_response_body_callback_fn)(
+    int error_code,
+    uint32_t num_failed,
+    uint32_t num_successful,
+    void *user_data);
+
 void aws_s3_client_stream_response_body(
     struct aws_s3_client *client,
     struct aws_s3_meta_request *meta_request,
-    struct aws_linked_list *requests);
+    struct aws_linked_list *requests,
+    aws_s3_client_stream_response_body_callback_fn callback,
+    void *user_data);
 
 AWS_EXTERN_C_BEGIN
 

include/aws/s3/private/s3_default_meta_request.h

@@ -0,0 +1,36 @@
+#ifndef AWS_S3_DEFAULT_META_REQUEST_H
+#define AWS_S3_DEFAULT_META_REQUEST_H
+
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#include "aws/s3/private/s3_meta_request_impl.h"
+
+struct aws_s3_client;
+
+struct aws_s3_meta_request_default {
+    struct aws_s3_meta_request base;
+
+    size_t content_length;
+
+    /* Members to only be used when the mutex in the base type is locked. */
+    struct {
+        int cached_response_status;
+        int request_error_code;
+
+        uint32_t request_sent : 1;
+        uint32_t request_completed : 1;
+
+    } synced_data;
+};
+
+/* Creates a new default meta request. This will send the request as is and pass back the response. */
+struct aws_s3_meta_request *aws_s3_meta_request_default_new(
+    struct aws_allocator *allocator,
+    struct aws_s3_client *client,
+    uint64_t content_length,
+    const struct aws_s3_meta_request_options *options);
+
+#endif