instead of skipping, let's prepare the parts and mark it noop

include/aws/s3/private/s3_auto_ranged_put.h

@@ -47,19 +47,6 @@ struct aws_s3_auto_ranged_put {
         uint32_t next_part_number;
     } threaded_update_data;
 
-    /*
-     * Should only be used during prepare requests. Note: stream reads must be sequential,
-     * so prepare currently never runs concurrently with another prepare
-     */
-    struct {
-        /*
-         * Start index of skipping parts.
-         * This is used to keep track of how many parts have been read from input steam and where to try to start
-         * skipping parts from.
-         */
-        uint32_t part_index_for_skipping;
-    } prepare_data;
-
     /* Members to only be used when the mutex in the base type is locked. */
     struct {
         /* Array list of `struct aws_s3_mpu_part_info *`

include/aws/s3/private/s3_meta_request_impl.h

@@ -138,9 +138,12 @@ struct aws_s3_meta_request {
     /* Initial HTTP Message that this meta request is based on. */
     struct aws_http_message *initial_request_message;
 
-    /* Async stream for meta request's body.
-     * NULL if using initial_request_message's synchronous body stream instead.  */
+    /* The meta request's outgoing body comes from one of these:
+     * 1) request_body_async_stream: if set, then async stream 1 part at a time
+     * 2) request_body_parallel_stream: if set, then stream multiple parts in parallel
+     * 3) initial_request_message's body_stream: else synchronously stream parts */
     struct aws_async_input_stream *request_body_async_stream;
+    struct aws_parallel_input_stream *request_body_parallel_stream;
 
     /* Part size to use for uploads and downloads.  Passed down by the creating client. */
     const size_t part_size;
@@ -259,6 +262,7 @@ struct aws_s3_mpu_part_info {
     uint64_t size;
     struct aws_string *etag;
     struct aws_byte_buf checksum_base64;
+    bool uploaded_before_resume;
 };
 
 AWS_EXTERN_C_BEGIN
@@ -363,10 +367,13 @@ bool aws_s3_meta_request_are_events_out_for_delivery_synced(struct aws_s3_meta_r
  * It may read from the underlying stream multiple times, if that's what it takes to fill the buffer.
  * Returns a future whose result bool indicates whether end of stream was reached.
  * This future may complete on any thread, and may complete synchronously.
+ *
+ * Read from offset to fill the buffer
  */
 AWS_S3_API
 struct aws_future_bool *aws_s3_meta_request_read_body(
     struct aws_s3_meta_request *meta_request,
+    uint64_t offset,
     struct aws_byte_buf *buffer);
 
 bool aws_s3_meta_request_body_has_no_more_data(const struct aws_s3_meta_request *meta_request);

include/aws/s3/private/s3_parallel_input_stream.h

@@ -0,0 +1,105 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#ifndef AWS_S3_PARALLEL_INPUT_STREAM_H
+#define AWS_S3_PARALLEL_INPUT_STREAM_H
+
+#include <aws/s3/s3.h>
+
+#include <aws/common/ref_count.h>
+
+AWS_PUSH_SANE_WARNING_LEVEL
+
+struct aws_byte_buf;
+struct aws_future_bool;
+struct aws_input_stream;
+
+struct aws_event_loop_group;
+
+struct aws_parallel_input_stream {
+    const struct aws_parallel_input_stream_vtable *vtable;
+    struct aws_allocator *alloc;
+    struct aws_ref_count ref_count;
+
+    void *impl;
+};
+
+struct aws_parallel_input_stream_vtable {
+    /**
+     * Destroy the stream, its refcount has reached 0.
+     */
+    void (*destroy)(struct aws_parallel_input_stream *stream);
+
+    /**
+     * Read into the buffer in parallel.
+     * The implementation needs to support this to be invoked concurrently from multiple threads
+     */
+    struct aws_future_bool *(
+        *read)(struct aws_parallel_input_stream *stream, uint64_t offset, struct aws_byte_buf *dest);
+};
+
+AWS_EXTERN_C_BEGIN
+
+/**
+ * Initialize aws_parallel_input_stream "base class"
+ */
+AWS_S3_API
+void aws_parallel_input_stream_init_base(
+    struct aws_parallel_input_stream *stream,
+    struct aws_allocator *alloc,
+    const struct aws_parallel_input_stream_vtable *vtable,
+    void *impl);
+
+/**
+ * Increment reference count.
+ * You may pass in NULL (has no effect).
+ * Returns whatever pointer was passed in.
+ */
+AWS_S3_API
+struct aws_parallel_input_stream *aws_parallel_input_stream_acquire(struct aws_parallel_input_stream *stream);
+
+/**
+ * Decrement reference count.
+ * You may pass in NULL (has no effect).
+ * Always returns NULL.
+ */
+AWS_S3_API
+struct aws_parallel_input_stream *aws_parallel_input_stream_release(struct aws_parallel_input_stream *stream);
+
+/**
+ * Read from the offset until fill the dest, or EOS reached.
+ * It's thread safe to be called from multiple threads without waiting for other read to complete
+ *
+ * @param stream            The stream to read from
+ * @param offset            The offset in the stream from beginning to start reading
+ * @param dest              The output buffer read to
+ * @return                  a future, which will contain an error code if something went wrong,
+ *                          or a result bool indicating whether EOF has been reached.
+ */
+AWS_S3_API
+struct aws_future_bool *aws_parallel_input_stream_read(
+    struct aws_parallel_input_stream *stream,
+    uint64_t offset,
+    struct aws_byte_buf *dest);
+
+/**
+ * Create a new file based parallel input stream implementation.
+ *
+ * This implementation will open a file handler when the read happens, and seek to the offset to start reading. Close
+ * the file handler as read finishes.
+ *
+ * @param allocator         memory allocator
+ * @param file_name         The file path to read from
+ * @return aws_parallel_input_stream
+ */
+AWS_S3_API
+struct aws_parallel_input_stream *aws_parallel_input_stream_new_from_file(
+    struct aws_allocator *allocator,
+    struct aws_byte_cursor file_name);
+
+AWS_EXTERN_C_END
+AWS_POP_SANE_WARNING_LEVEL
+
+#endif /* AWS_S3_PARALLEL_INPUT_STREAM_H */

include/aws/s3/private/s3_request.h

@@ -202,6 +202,9 @@ struct aws_s3_request {
      * requests for uploading data after the end of the stream (those requests
      * will use below flag to indicate that they should not be sent). */
     uint32_t is_noop : 1;
+
+    /* When true, this request has already been uploaded. we still prepare the request to check the durability. */
+    uint32_t uploaded_before_resume : 1;
 };
 
 AWS_EXTERN_C_BEGIN

include/aws/s3/private/s3_request_messages.h

@@ -37,13 +37,6 @@ struct aws_http_message *aws_s3_message_util_copy_http_message_no_body_filter_he
     size_t excluded_headers_size,
     bool exclude_x_amz_meta);
 
-/* Copy message and retain all headers, but replace body with one that reads directly from a filepath. */
-AWS_S3_API
-struct aws_http_message *aws_s3_message_util_copy_http_message_filepath_body_all_headers(
-    struct aws_allocator *allocator,
-    struct aws_http_message *message,
-    struct aws_byte_cursor filepath);
-
 /* Copy headers from one message to the other and exclude specific headers.
  * exclude_x_amz_meta controls whether S3 user metadata headers (prefixed with "x-amz-meta) are excluded.*/
 AWS_S3_API

include/aws/s3/s3.h

@@ -40,6 +40,7 @@ enum aws_s3_errors {
     AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE,
     AWS_ERROR_S3_INCORRECT_CONTENT_LENGTH,
     AWS_ERROR_S3_REQUEST_TIME_TOO_SKEWED,
+    AWS_ERROR_S3_FILE_MODIFIED,
     AWS_ERROR_S3_END_RANGE = AWS_ERROR_ENUM_END_RANGE(AWS_C_S3_PACKAGE_ID)
 };
 

source/s3.c

@@ -39,6 +39,7 @@ static struct aws_error_info s_errors[] = {
     AWS_DEFINE_ERROR_INFO_S3(AWS_ERROR_S3_METRIC_DATA_NOT_AVAILABLE, "The metric data is not available, the requests ends before the metric happens."),
     AWS_DEFINE_ERROR_INFO_S3(AWS_ERROR_S3_INCORRECT_CONTENT_LENGTH, "Request body length must match Content-Length header."),
     AWS_DEFINE_ERROR_INFO_S3(AWS_ERROR_S3_REQUEST_TIME_TOO_SKEWED, "RequestTimeTooSkewed error received from S3."),
+    AWS_DEFINE_ERROR_INFO_S3(AWS_ERROR_S3_FILE_MODIFIED, "The file was modified during upload."),
 };
 /* clang-format on */