couple fixes

TingDaoK · TingDaoK · commit 9303470235ce · 2025-07-15T15:17:58.000-07:00
diff --git a/include/aws/s3/private/s3_meta_request_impl.h b/include/aws/s3/private/s3_meta_request_impl.h
@@ -25,7 +25,6 @@ struct aws_s3_request;
 struct aws_http_headers;
 struct aws_http_make_request_options;
 struct aws_retry_strategy;
-struct aws_mmap_context;
 
 enum aws_s3_meta_request_state {
     AWS_S3_META_REQUEST_STATE_ACTIVE,
diff --git a/include/aws/s3/private/s3_parallel_input_stream.h b/include/aws/s3/private/s3_parallel_input_stream.h
@@ -33,13 +33,20 @@ struct aws_parallel_input_stream_vtable {
     void (*destroy)(struct aws_parallel_input_stream *stream);
 
     /**
-     * Read into the buffer in parallel.
-     * The stream will split into parts and read each part in parallel.
+     * 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,
-        size_t offset,
-        size_t length,
+        uint64_t offset,
+        size_t max_length,
         struct aws_byte_buf *dest);
 };
 
@@ -72,20 +79,21 @@ AWS_S3_API
 struct aws_parallel_input_stream *aws_parallel_input_stream_release(struct aws_parallel_input_stream *stream);
 
 /**
- * Read data from the stream into the destination buffer.
+ * 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 parallel input stream to read from
- * @param offset The starting offset in the stream to read from
- * @param length The number of bytes to read
- * @param dest The destination buffer to read into
- * @return A future that will be set to true when the read completes successfully,
- *         or set to an error if the read fails
+ * @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.
  */
 AWS_S3_API
 struct aws_future_bool *aws_parallel_input_stream_read(
     struct aws_parallel_input_stream *stream,
-    size_t offset,
-    size_t length,
+    uint64_t offset,
+    size_t max_length,
     struct aws_byte_buf *dest);
 
 /**
diff --git a/source/s3_auto_ranged_put.c b/source/s3_auto_ranged_put.c
@@ -571,7 +571,9 @@ static bool s_s3_auto_ranged_put_update(
 
                 /* Allocate a request for another part. */
                 uint32_t new_flags = AWS_S3_REQUEST_FLAG_RECORD_RESPONSE_HEADERS;
-                if (!meta_request->synced_data.async_write.ready_to_send) {
+                if (!meta_request->synced_data.async_write.ready_to_send &&
+                    meta_request->request_body_parallel_stream == NULL) {
+                    /* TODO: now get around the memory pool when streaming from file, but we can still hook it up. */
                     new_flags |= AWS_S3_REQUEST_FLAG_ALLOCATE_BUFFER_FROM_POOL;
                 }
 
diff --git a/source/s3_meta_request.c b/source/s3_meta_request.c
@@ -2248,9 +2248,10 @@ struct aws_future_bool *aws_s3_meta_request_read_body(
     }
 
     /* If parallel-stream, simply call read(), which must fill the buffer and/or EOF */
-    // if (meta_request->request_body_parallel_stream != NULL) {
-    //     return aws_parallel_input_stream_read(meta_request->request_body_parallel_stream, offset, buffer);
-    // }
+    if (meta_request->request_body_parallel_stream != NULL) {
+        return aws_parallel_input_stream_read(
+            meta_request->request_body_parallel_stream, offset, buffer->capacity, buffer);
+    }
 
     /* Further techniques are synchronous... */
     struct aws_future_bool *synchronous_read_future = aws_future_bool_new(meta_request->allocator);
diff --git a/source/s3_parallel_input_stream.c b/source/s3_parallel_input_stream.c
@@ -15,6 +15,8 @@
 #include <aws/io/future.h>
 #include <aws/io/stream.h>
 
+#include <errno.h>
+
 #define ONE_SEC_IN_NS_P ((uint64_t)AWS_TIMESTAMP_NANOS)
 #define MAX_TIMEOUT_NS_P (600 * ONE_SEC_IN_NS_P)
 
@@ -47,8 +49,8 @@ struct aws_parallel_input_stream *aws_parallel_input_stream_release(struct aws_p
 
 struct aws_future_bool *aws_parallel_input_stream_read(
     struct aws_parallel_input_stream *stream,
-    size_t offset,
-    size_t length,
+    uint64_t offset,
+    size_t max_length,
     struct aws_byte_buf *dest) {
     /* Ensure the buffer has space available */
     if (dest->len == dest->capacity) {
@@ -57,7 +59,7 @@ struct aws_future_bool *aws_parallel_input_stream_read(
         return future;
     }
 
-    struct aws_future_bool *future = stream->vtable->read(stream, offset, length, dest);
+    struct aws_future_bool *future = stream->vtable->read(stream, offset, max_length, dest);
     AWS_POSTCONDITION(future != NULL);
     return future;
 }
@@ -86,7 +88,7 @@ struct read_task_impl {
     struct aws_parallel_input_stream_from_file_impl *para_impl;
 
     struct aws_future_bool *end_future;
-    size_t offset;
+    uint64_t offset;
     size_t length;
     struct aws_byte_buf *dest;
 };
@@ -98,13 +100,13 @@ static void s_s3_parallel_from_file_read_task(struct aws_task *task, void *arg,
     struct aws_future_bool *end_future = read_task->end_future;
     FILE *file_stream = NULL;
     int error_code = AWS_ERROR_SUCCESS;
-    
+
     file_stream = aws_fopen(aws_string_c_str(impl->file_path), "rb");
     if (file_stream == NULL) {
         AWS_LOGF_ERROR(
-            AWS_LS_S3_GENERAL, 
-            "id=%p: Failed to open file %s for reading", 
-            (void *)&impl->base, 
+            AWS_LS_S3_GENERAL,
+            "id=%p: Failed to open file %s for reading",
+            (void *)&impl->base,
             aws_string_c_str(impl->file_path));
         error_code = aws_last_error();
         goto cleanup;
@@ -113,10 +115,10 @@ static void s_s3_parallel_from_file_read_task(struct aws_task *task, void *arg,
     /* seek to the right position and then read */
     if (aws_fseek(file_stream, (int64_t)read_task->offset, SEEK_SET)) {
         AWS_LOGF_ERROR(
-            AWS_LS_S3_GENERAL, 
-            "id=%p: Failed to seek to position %zu in file %s", 
-            (void *)&impl->base, 
-            read_task->offset, 
+            AWS_LS_S3_GENERAL,
+            "id=%p: Failed to seek to position %llu in file %s",
+            (void *)&impl->base,
+            (unsigned long long)read_task->offset,
             aws_string_c_str(impl->file_path));
         error_code = aws_last_error();
         goto cleanup;
@@ -125,66 +127,71 @@ static void s_s3_parallel_from_file_read_task(struct aws_task *task, void *arg,
     size_t actually_read = fread(read_task->dest->buffer + read_task->dest->len, 1, read_task->length, file_stream);
     if (actually_read == 0 && ferror(file_stream)) {
         AWS_LOGF_ERROR(
-            AWS_LS_S3_GENERAL, 
-            "id=%p: Failed to read %zu bytes from file %s", 
-            (void *)&impl->base, 
-            read_task->length, 
+            AWS_LS_S3_GENERAL,
+            "id=%p: Failed to read %zu bytes from file %s",
+            (void *)&impl->base,
+            read_task->length,
             aws_string_c_str(impl->file_path));
-        error_code = aws_last_error();
+        error_code = aws_translate_and_raise_io_error(errno);
         goto cleanup;
     }
-    
+
     read_task->dest->len += actually_read;
-    
+
     AWS_LOGF_TRACE(
-        AWS_LS_S3_GENERAL, 
-        "id=%p: Successfully read %zu bytes from file %s at position %zu", 
-        (void *)&impl->base, 
-        actually_read, 
-        aws_string_c_str(impl->file_path), 
-        read_task->offset);
+        AWS_LS_S3_GENERAL,
+        "id=%p: Successfully read %zu bytes from file %s at position %llu",
+        (void *)&impl->base,
+        actually_read,
+        aws_string_c_str(impl->file_path),
+        (unsigned long long)read_task->offset);
 
 cleanup:
     if (file_stream != NULL) {
         fclose(file_stream);
     }
-    
+
     if (error_code != AWS_ERROR_SUCCESS) {
         aws_future_bool_set_error(end_future, error_code);
     } else {
-        aws_future_bool_set_result(end_future, true);
+        /* Return true if we reached EOF */
+        bool eof_reached = (actually_read < read_task->length);
+        aws_future_bool_set_result(end_future, eof_reached);
     }
-    
+
     aws_future_bool_release(end_future);
     aws_mem_release(impl->base.alloc, task);
     aws_mem_release(impl->base.alloc, read_task);
 }
 
 struct aws_future_bool *s_para_from_file_read(
     struct aws_parallel_input_stream *stream,
-    size_t offset,
-    size_t length,
+    uint64_t offset,
+    size_t max_length,
     struct aws_byte_buf *dest) {
 
     struct aws_future_bool *future = aws_future_bool_new(stream->alloc);
     struct aws_parallel_input_stream_from_file_impl *impl =
         AWS_CONTAINER_OF(stream, struct aws_parallel_input_stream_from_file_impl, base);
 
-    if (!length) {
+    /* Calculate how much we can read based on available buffer space and max_length */
+    size_t available_space = dest->capacity - dest->len;
+    size_t length = aws_min_size(available_space, max_length);
+
+    if (length == 0) {
         /* Nothing to read. Complete the read with success. */
-        aws_future_bool_set_result(future, true);
-        return future;
-    }
-    
-    if (length > dest->capacity - dest->len) {
-        AWS_LOGF_ERROR(AWS_LS_S3_GENERAL, "id=%p: The buffer read to cannot fit the data.", (void *)stream);
-        aws_future_bool_set_error(future, AWS_ERROR_SHORT_BUFFER);
+        aws_future_bool_set_result(future, false);
         return future;
     }
 
     struct read_task_impl *read_task = aws_mem_calloc(impl->base.alloc, 1, sizeof(struct read_task_impl));
 
-    AWS_LOGF_TRACE(AWS_LS_S3_GENERAL, "id=%p: Read %zu bytes from offset %zu", (void *)stream, length, offset);
+    AWS_LOGF_TRACE(
+        AWS_LS_S3_GENERAL,
+        "id=%p: Read %zu bytes from offset %llu",
+        (void *)stream,
+        length,
+        (unsigned long long)offset);
 
     /* Initialize for one read */
     read_task->dest = dest;
@@ -213,9 +220,17 @@ struct aws_parallel_input_stream *aws_parallel_input_stream_new_from_file(
 
     struct aws_parallel_input_stream_from_file_impl *impl =
         aws_mem_calloc(allocator, 1, sizeof(struct aws_parallel_input_stream_from_file_impl));
+
+    aws_parallel_input_stream_init_base(&impl->base, allocator, &s_parallel_input_stream_from_file_vtable, impl);
     impl->file_path = aws_string_new_from_cursor(allocator, &file_name);
     impl->reading_elg = aws_event_loop_group_acquire(reading_elg);
-    aws_parallel_input_stream_init_base(&impl->base, allocator, &s_parallel_input_stream_from_file_vtable, impl);
+
+    if (!aws_path_exists(impl->file_path)) {
+        /* If file path not exists, raise error from errno. */
+        aws_translate_and_raise_io_error(errno);
+        s_para_from_file_destroy(&impl->base);
+        return NULL;
+    }
 
     return &impl->base;
 }
@@ -262,6 +277,12 @@ static int s_aws_s3_mmap_part_streaming_input_stream_read(struct aws_input_strea
         /* The reading buf is invalid. Block until the loading buf is available. */
         AWS_ASSERT(impl->loading_future != NULL);
         aws_future_bool_wait(impl->loading_future, MAX_TIMEOUT_NS_P);
+        int read_error = aws_future_bool_get_error(impl->loading_future);
+        if (read_error != 0) {
+            /* Read failed. */
+            return aws_raise_error(read_error);
+        }
+        bool eos = aws_future_bool_get_result(impl->loading_future);
         impl->loading_future = aws_future_bool_release(impl->loading_future);
         /* Swap the reading the loading pointer. */
         AWS_ASSERT(impl->reading_chunk_buf->len == 0);
@@ -271,9 +292,11 @@ static int s_aws_s3_mmap_part_streaming_input_stream_read(struct aws_input_strea
         size_t new_offset = impl->offset + impl->total_length_read + impl->chunk_load_size;
         size_t new_load_length = aws_min_size(
             impl->chunk_load_size, impl->total_length - impl->total_length_read - impl->reading_chunk_buf->len);
-        /* Kick off loading the next chunk. */
-        impl->loading_future = aws_parallel_input_stream_read(
-            impl->stream, new_offset, new_load_length, impl->loading_chunk_buf);
+        if (new_load_length > 0 && !eos) {
+            /* Kick off loading the next chunk. */
+            impl->loading_future =
+                aws_parallel_input_stream_read(impl->stream, new_offset, new_load_length, impl->loading_chunk_buf);
+        }
         impl->in_chunk_offset = 0;
     }
     read_length = aws_min_size(read_length, impl->reading_chunk_buf->len - impl->in_chunk_offset);
diff --git a/source/s3_request.c b/source/s3_request.c
@@ -37,9 +37,7 @@ struct aws_s3_request *aws_s3_request_new(
 
     request->part_number = part_number;
     request->record_response_headers = (flags & AWS_S3_REQUEST_FLAG_RECORD_RESPONSE_HEADERS) != 0;
-    /* TODO: if the request is streaming, we are not allocating buffer from pool/ This will avoid the allocation buffer
-     * to be the bottleneck when streaming. */
-    request->should_allocate_buffer_from_pool = false;
+    request->should_allocate_buffer_from_pool = (flags & AWS_S3_REQUEST_FLAG_ALLOCATE_BUFFER_FROM_POOL) != 0;
     request->always_send = (flags & AWS_S3_REQUEST_FLAG_ALWAYS_SEND) != 0;
 
     return request;
diff --git a/source/s3express_credentials_provider.c b/source/s3express_credentials_provider.c
@@ -26,7 +26,11 @@ static struct aws_byte_cursor s_create_session_path_query = AWS_BYTE_CUR_INIT_FR
 static const size_t s_default_cache_capacity = 100;
 
 /* Those number are from C++ SDK impl */
-static const uint64_t s_expired_threshold_secs = 60;
+/**
+ * TODO: When each part is huge, the 5 sec is probably causing the credentials to be expired. Consider moving the
+ * signing closer to HTTP request.
+ */
+static const uint64_t s_expired_threshold_secs = 5;
 static const uint64_t s_about_to_expire_threshold_secs = 60;
 static const uint64_t s_background_refresh_interval_secs = 60;
 
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
@@ -384,8 +384,8 @@ if(AWS_ENABLE_S3_ENDPOINT_RESOLVER)
     add_test_case(test_s3_endpoint_resolver_resolve_endpoint_force_path_style)
 endif()
 
-# add_test_case(parallel_read_stream_from_file_sanity_test)
-# add_test_case(parallel_read_stream_from_large_file_test)
+add_test_case(parallel_read_stream_from_file_sanity_test)
+add_test_case(parallel_read_stream_from_large_file_test)
 
 add_test_case(test_s3_buffer_pool_threaded_allocs_and_frees)
 add_test_case(test_s3_buffer_pool_large_chunk_threaded_allocs_and_frees)
diff --git a/tests/s3_data_plane_tests.c b/tests/s3_data_plane_tests.c
diff --git a/tests/s3_parallel_read_stream_test.c b/tests/s3_parallel_read_stream_test.c
diff --git a/tests/s3_test_parallel_stream.c b/tests/s3_test_parallel_stream.c

Original file line number	Diff line number	Diff line change
`@@ -571,7 +571,9 @@ static bool s_s3_auto_ranged_put_update(`
`571`	`571`
`572`	`572`	`/* Allocate a request for another part. */`
`573`	`573`	`uint32_t new_flags = AWS_S3_REQUEST_FLAG_RECORD_RESPONSE_HEADERS;`
`574`		`- if (!meta_request->synced_data.async_write.ready_to_send) {`
	`574`	`+ if (!meta_request->synced_data.async_write.ready_to_send &&`
	`575`	`+ meta_request->request_body_parallel_stream == NULL) {`
	`576`	`+ /* TODO: now get around the memory pool when streaming from file, but we can still hook it up. */`
`575`	`577`	`new_flags \|= AWS_S3_REQUEST_FLAG_ALLOCATE_BUFFER_FROM_POOL;`
`576`	`578`	`}`
`577`	`579`