File streaming support (#564)

- Support streaming from file for upload instead of buffer the full part
reviewed in 5 separate PRs:
- #556
- #549
- #546
- #544
- #566

Author: TingDaoK

include/aws/s3/private/s3_checksums.h

@@ -62,7 +62,29 @@ struct aws_s3_meta_request_checksum_config_storage {
 };
 
 /**
- * a stream that takes in a stream
+ * Helper stream that takes in a stream and the checksum context to help finalize the checksum from the underlying
+ * stream.
+ * The context will be only finalized when the checksum stream has read to the end of stream.
+ *
+ * Note: seek this stream will immediately fail, as it would prevent an accurate calculation of the
+ * checksum.
+ *
+ * @param allocator
+ * @param existing_stream The real content to read from. Destroying the checksum stream destroys the existing stream.
+ *                        outputs the checksum of existing stream to checksum_output upon destruction. Will be kept
+ *                        alive by the checksum stream
+ * @param context         Checksum context to keep and get checksum requirements from.
+ */
+AWS_S3_API
+struct aws_input_stream *aws_checksum_stream_new_with_context(
+    struct aws_allocator *allocator,
+    struct aws_input_stream *existing_stream,
+    struct aws_s3_upload_request_checksum_context *context);
+
+/**
+ * Helper stream that takes in a stream to keep track of the checksum of the underlying stream during read.
+ * Invoke `aws_checksum_stream_finalize_checksum` to get the checksum of the data has been read so far.
+ *
  * Note: seek this stream will immediately fail, as it would prevent an accurate calculation of the
  * checksum.
  *
@@ -85,15 +107,6 @@ struct aws_input_stream *aws_checksum_stream_new(
 AWS_S3_API
 int aws_checksum_stream_finalize_checksum(struct aws_input_stream *checksum_stream, struct aws_byte_buf *checksum_buf);
 
-/**
- * Finalize the checksum has read so far to the checksum context.
- * Not thread safe.
- */
-AWS_S3_API
-int aws_checksum_stream_finalize_checksum_context(
-    struct aws_input_stream *checksum_stream,
-    struct aws_s3_upload_request_checksum_context *checksum_context);
-
 /**
  * TODO: properly support chunked encoding.
  * Creates a chunked encoding stream that wraps an existing stream and adds checksum trailers.

include/aws/s3/private/s3_client_impl.h

@@ -168,8 +168,11 @@ struct aws_s3_client_vtable {
 
     void (*finish_destroy)(struct aws_s3_client *client);
 
-    struct aws_parallel_input_stream *(
-        *parallel_input_stream_new_from_file)(struct aws_allocator *allocator, struct aws_byte_cursor file_name);
+    struct aws_parallel_input_stream *(*parallel_input_stream_new_from_file)(
+        struct aws_allocator *allocator,
+        struct aws_byte_cursor file_name,
+        struct aws_event_loop_group *reading_elg,
+        bool direct_io_read);
 
     struct aws_http_stream *(*http_connection_make_request)(
         struct aws_http_connection *client_connection,
@@ -235,6 +238,10 @@ struct aws_s3_client {
      * to meta requests for use. */
     const uint64_t max_part_size;
 
+    /* File I/O options. */
+    bool fio_options_set;
+    struct aws_s3_file_io_options fio_opts;
+
     /* The size threshold in bytes for when to use multipart uploads for a AWS_S3_META_REQUEST_TYPE_PUT_OBJECT meta
      * request. Uploads over this size will automatically use a multipart upload strategy, while uploads smaller or
      * equal to this threshold will use a single request to upload the whole object. If not set, `part_size` will be
@@ -351,6 +358,9 @@ struct aws_s3_client {
 
         /* Number of requests currently scheduled to be streamed the response body or are actively being streamed. */
         struct aws_atomic_var num_requests_streaming_response;
+
+        /* Number of overall requests currently streaming the request body instead of buffering. */
+        struct aws_atomic_var num_requests_streaming_request_body;
     } stats;
 
     struct {

include/aws/s3/private/s3_meta_request_impl.h

@@ -288,6 +288,9 @@ struct aws_s3_meta_request {
     FILE *recv_file;
     struct aws_string *recv_filepath;
     bool recv_file_delete_on_failure;
+
+    /* File I/O options. */
+    struct aws_s3_file_io_options fio_opts;
 };
 
 /* Info for each part, that we need to remember until we send CompleteMultipartUpload */
@@ -307,6 +310,7 @@ int aws_s3_meta_request_init_base(
     struct aws_s3_client *client,
     size_t part_size,
     bool should_compute_content_md5,
+    bool should_default_streaming,
     const struct aws_s3_meta_request_options *options,
     void *impl,
     struct aws_s3_meta_request_vtable *vtable,

include/aws/s3/private/s3_parallel_input_stream.h

@@ -14,14 +14,21 @@ AWS_PUSH_SANE_WARNING_LEVEL
 
 struct aws_byte_buf;
 struct aws_future_bool;
+struct aws_future_void;
 struct aws_input_stream;
+struct aws_s3_meta_request;
+struct aws_s3_request;
 
 struct aws_event_loop_group;
 
+/**
+ * This should be private, but keep it public for providing your own implementation.
+ */
 struct aws_parallel_input_stream {
     const struct aws_parallel_input_stream_vtable *vtable;
     struct aws_allocator *alloc;
     struct aws_ref_count ref_count;
+    struct aws_future_void *shutdown_future;
 
     void *impl;
 };
@@ -33,11 +40,27 @@ struct aws_parallel_input_stream_vtable {
     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
+     * Read from the offset until fill the dest, or EOF 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 max_length        The maximum number of bytes to read
+     * @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.
      */
     struct aws_future_bool *(
-        *read)(struct aws_parallel_input_stream *stream, uint64_t offset, struct aws_byte_buf *dest);
+        *read)(struct aws_parallel_input_stream *stream, uint64_t offset, size_t max_length, struct aws_byte_buf *dest);
+
+    /**
+     * Get the length of the stream.
+     *
+     * @param stream            The stream to get length from
+     * @param out_length        The output length
+     * @return                  AWS_OP_SUCCESS if success, otherwise AWS_OP_ERR
+     */
+    int (*get_length)(struct aws_parallel_input_stream *stream, int64_t *out_length);
 };
 
 AWS_EXTERN_C_BEGIN
@@ -74,6 +97,7 @@ struct aws_parallel_input_stream *aws_parallel_input_stream_release(struct aws_p
  *
  * @param stream            The stream to read from
  * @param offset            The offset in the stream from beginning to start reading
+ * @param max_length        The maximum number of bytes to read
  * @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.
@@ -82,22 +106,51 @@ AWS_S3_API
 struct aws_future_bool *aws_parallel_input_stream_read(
     struct aws_parallel_input_stream *stream,
     uint64_t offset,
+    size_t max_length,
     struct aws_byte_buf *dest);
 
 /**
- * Create a new file based parallel input stream.
+ * Get the total length of the parallel input stream.
  *
- * 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 stream
+ * @param out_length
+ * @return AWS_S3_API
+ */
+AWS_S3_API
+int aws_parallel_input_stream_get_length(struct aws_parallel_input_stream *stream, int64_t *out_length);
+
+/**
+ * Creates a new parallel input stream that reads from a file.
+ * This stream uses an event loop group to perform file I/O operations asynchronously.
+ *
+ * Notes for direct_io_read:
+ * - checking `aws_file_path_read_from_offset_direct_io` for detail
+ * - For `AWS_ERROR_UNSUPPORTED_OPERATION`, fallback to reading with cache with warnings, instead of fail.
+ * - If alignment required, it's callers' responsibility to align with the page size.
  *
- * @param allocator         memory allocator
- * @param file_name         The file path to read from
- * @return aws_parallel_input_stream
+ * @param allocator The allocator to use for memory allocation
+ * @param file_name The name of the file to read from
+ * @param reading_elg The event loop group to use for file I/O operations
+ * @param direct_io_read Whether to use direct I/O for reading the file.
+ *
+ * @return A new parallel input stream that reads from the specified file
  */
 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);
+    struct aws_byte_cursor file_name,
+    struct aws_event_loop_group *reading_elg,
+    bool direct_io_read);
+
+/**
+ * Get the shutdown future from the parallel input stream.
+ * The future will be completed when every refcount on the stream has been released.
+ * And all the resource has been released.
+ * Don't hold any refcount of the stream while waiting on the future, otherwise, deadlock can happen.
+ * You need to release the future after using it.
+ */
+AWS_S3_API
+struct aws_future_void *aws_parallel_input_stream_get_shutdown_future(struct aws_parallel_input_stream *stream);
 
 AWS_EXTERN_C_END
 AWS_POP_SANE_WARNING_LEVEL

include/aws/s3/private/s3_part_streaming_input_stream.h

@@ -0,0 +1,48 @@
+/**
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0.
+ */
+
+#ifndef AWS_S3_PART_STREAMING_INPUT_STREAM_H
+#define AWS_S3_PART_STREAMING_INPUT_STREAM_H
+
+#include <aws/s3/s3.h>
+
+struct aws_input_stream;
+struct aws_s3_buffer_ticket;
+struct aws_parallel_input_stream;
+
+AWS_PUSH_SANE_WARNING_LEVEL
+AWS_EXTERN_C_BEGIN
+
+/**
+ * Creates a new streaming input stream that reads from a parallel input stream.
+ * This adapter allows using a parallel input stream with APIs that expect a standard input stream.
+ * The adapter uses double-buffering to read ahead and provide efficient streaming.
+ *
+ * Note: The input stream only provides a blocking API to read, if reading from the same thread that
+ *      will be used from `para_stream` to handle the parallel read, it's subjected to a dead lock!!
+ *      Make sure the input stream is read from a different thread/thread pool that executes the `para_stream` reads.
+ *
+ * @param allocator             The allocator to use for memory allocation
+ * @param para_stream           The parallel input stream to read from
+ * @param buffer_ticket         The buffer pool ticket to use for buffering
+ * @param offset                The starting offset in the stream
+ * @param request_body_size     The total size to read
+ * @param page_aligned          Whether the input stream only read from the para_stream on that
+ *                              aligned with the page size, required for direct I/O in linux.
+ * @return A new input stream that reads from the parallel input stream
+ */
+AWS_S3_API
+struct aws_input_stream *aws_part_streaming_input_stream_new(
+    struct aws_allocator *allocator,
+    struct aws_parallel_input_stream *para_stream,
+    struct aws_s3_buffer_ticket *buffer_ticket,
+    uint64_t offset,
+    size_t request_body_size,
+    bool page_aligned);
+
+AWS_EXTERN_C_END
+AWS_POP_SANE_WARNING_LEVEL
+
+#endif /* AWS_S3_PART_STREAMING_INPUT_STREAM_H */

include/aws/s3/private/s3_request.h

@@ -157,10 +157,19 @@ struct aws_s3_request {
     /* Owning meta request. */
     struct aws_s3_meta_request *meta_request;
 
+    /* The buffer size to be allocated for the request, defaults the part size. */
+    size_t buffer_size;
     /* Request body to use when sending the request. The contents of this body will be re-used if a request is
      * retried.*/
     struct aws_byte_buf request_body;
 
+    /* Set when the request will be streaming from file directly instead of the request_body. */
+    bool fio_streaming;
+    /* If file I/O options configure streaming. The request body will be streaming from this. If a request is retried,
+     * this stream will be recreated. */
+    struct aws_input_stream *request_body_stream;
+    uint64_t content_length;
+
     /**
      * Ticket to acquire the buffer.
      */

include/aws/s3/private/s3_request_messages.h

@@ -52,6 +52,7 @@ AWS_S3_API
 struct aws_input_stream *aws_s3_message_util_assign_body(
     struct aws_allocator *allocator,
     struct aws_byte_buf *byte_buf,
+    struct aws_input_stream *stream,
     struct aws_http_message *out_message,
     struct aws_s3_upload_request_checksum_context *checksum_context);
 
@@ -91,6 +92,18 @@ struct aws_http_message *aws_s3_upload_part_message_new(
     bool should_compute_content_md5,
     struct aws_s3_upload_request_checksum_context *checksum_context);
 
+/* Same as `aws_s3_upload_part_message_new`, but instead of taking the loaded buffer, taking the input stream for
+ * streaming I/O. */
+AWS_S3_API
+struct aws_http_message *aws_s3_upload_part_message_new_streaming(
+    struct aws_allocator *allocator,
+    struct aws_http_message *base_message,
+    struct aws_input_stream *input_stream,
+    uint32_t part_number,
+    const struct aws_string *upload_id,
+    bool should_compute_content_md5,
+    struct aws_s3_upload_request_checksum_context *checksum_context);
+
 /* Create an HTTP request for an S3 UploadPartCopy request, using the original request as a basis.
  * If multipart is not needed, part number and upload_id can be 0 and NULL,
  * respectively. */

include/aws/s3/private/s3_util.h

@@ -24,7 +24,8 @@
 #endif
 #define KB_TO_BYTES(kb) ((kb) * 1024)
 #define MB_TO_BYTES(mb) ((mb) * 1024 * 1024)
-#define GB_TO_BYTES(gb) ((gb) * 1024 * 1024 * 1024ULL)
+#define GB_TO_BYTES(gb) ((uint64_t)(gb) * 1024 * 1024 * 1024ULL)
+#define TB_TO_BYTES(tb) ((uint64_t)(tb) * 1024 * 1024 * 1024 * 1024ULL)
 
 #define MS_TO_NS(ms) ((uint64_t)(ms) * 1000000)
 #define SEC_TO_NS(ms) ((uint64_t)(ms) * 1000000000)
@@ -150,6 +151,15 @@ extern const struct aws_byte_cursor g_delete_method;
 AWS_S3_API
 extern const uint32_t g_s3_max_num_upload_parts;
 
+AWS_S3_API
+/* TODO: now this is hard-coded as 8MB, but maybe something else is better. */
+extern const size_t g_streaming_buffer_size;
+
+AWS_S3_API
+extern const double g_default_throughput_target_gbps;
+
+AWS_S3_API
+extern const uint64_t g_streaming_object_size_threshold;
 /**
  * Returns AWS_S3_REQUEST_TYPE_UNKNOWN if name doesn't map to an enum value.
  */