collect metrics

Author: TingDaoK

include/aws/s3/private/s3_meta_request_impl.h

@@ -124,6 +124,13 @@ struct aws_s3_meta_request_vtable {
     int (*pause)(struct aws_s3_meta_request *meta_request, struct aws_s3_meta_request_resume_token **resume_token);
 };
 
+struct s3_data_read_metrics {
+    uint64_t offset;
+    uint64_t size;
+    uint64_t start_timestamp;
+    uint64_t end_timestamp;
+};
+
 /**
  * This represents one meta request, ie, one accelerated file transfer.  One S3 meta request can represent multiple S3
  * requests.
@@ -139,6 +146,7 @@ struct aws_s3_meta_request {
 
     /* Initial HTTP Message that this meta request is based on. */
     struct aws_http_message *initial_request_message;
+    struct aws_array_list read_metrics_list;
 
     /* 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

include/aws/s3/private/s3_parallel_input_stream.h

@@ -124,6 +124,7 @@ struct aws_parallel_input_stream *aws_parallel_input_stream_new_from_file(
 struct aws_input_stream *aws_input_stream_new_from_parallel_stream(
     struct aws_allocator *allocator,
     struct aws_parallel_input_stream *stream,
+    struct aws_s3_meta_request *meta_request,
     uint64_t offset,
     size_t request_body_size);
 

source/s3_auto_ranged_put.c

@@ -982,7 +982,7 @@ struct aws_future_http_message *s_s3_prepare_upload_part(struct aws_s3_request *
             uint64_t offset = 0;
             size_t request_body_size = s_compute_request_body_size(meta_request, request->part_number, &offset);
             request->request_stream = aws_input_stream_new_from_parallel_stream(
-                allocator, meta_request->request_body_parallel_stream, offset, request_body_size);
+                allocator, meta_request->request_body_parallel_stream, meta_request, offset, request_body_size);
             request->content_length = request_body_size;
             struct aws_s3_auto_ranged_put *auto_ranged_put = meta_request->impl;
 
@@ -1020,14 +1020,71 @@ struct aws_future_http_message *s_s3_prepare_upload_part(struct aws_s3_request *
             s_s3_prepare_upload_part_finish(part_prep, AWS_ERROR_SUCCESS);
         } else {
             printf("PARALLEL retry 8MB read\n");
+            // Create unique filename with timestamp and meta_request pointer
+            char filename[256];
+            struct timespec ts;
+            clock_gettime(CLOCK_REALTIME, &ts);
+            snprintf(
+                filename,
+                sizeof(filename),
+                "/tmp/s3_read_metrics_%p_%ld_%ld.csv",
+                (void *)meta_request,
+                ts.tv_sec,
+                ts.tv_nsec);
+
+            FILE *metrics_file = fopen(filename, "w");
+            /* BEGIN CRITICAL SECTION */
+            aws_s3_meta_request_lock_synced_data(meta_request);
+            /* write every read metric to a file */
+            size_t metric_length = aws_array_list_length(&meta_request->read_metrics_list);
+            /* write every read metric to a file */
+            if (metrics_file) {
+                // Write CSV header
+                fprintf(metrics_file, "index,offset,size,start_timestamp,end_timestamp,duration_ns,throughput_mbps\n");
+                // Write all metrics
+                for (size_t j = 0; j < metric_length; j++) {
+                    struct s3_data_read_metrics m;
+                    aws_array_list_get_at(&meta_request->read_metrics_list, &m, j);
+
+                    uint64_t duration = m.end_timestamp - m.start_timestamp;
+                    double throughput_mbps =
+                        duration > 0 ? (double)(m.size * 8) / (duration / 1000.0) / 1000000.0 : 0.0;
+
+                    fprintf(
+                        metrics_file,
+                        "%zu,%llu,%llu,%llu,%llu,%llu,%.2f\n",
+                        j,
+                        (unsigned long long)m.offset,
+                        (unsigned long long)m.size,
+                        (unsigned long long)m.start_timestamp,
+                        (unsigned long long)m.end_timestamp,
+                        (unsigned long long)duration,
+                        throughput_mbps);
+                }
+                aws_array_list_clean_up(&meta_request->read_metrics_list);
+                fclose(metrics_file);
+
+                AWS_LOGF_INFO(
+                    AWS_LS_S3_META_REQUEST,
+                    "id=%p Wrote %zu read metrics to %s",
+                    (void *)meta_request,
+                    metric_length,
+                    filename);
+            } else {
+                AWS_LOGF_ERROR(
+                    AWS_LS_S3_META_REQUEST, "id=%p Failed to open metrics file %s", (void *)meta_request, filename);
+            }
+            aws_s3_meta_request_unlock_synced_data(meta_request);
+            /* END CRITICAL SECTION */
+
             /* Not the first time preparing request (e.g. retry).
              * We can skip over the async steps that read the body stream */
             /* Seek back to beginning of the stream. */
             aws_input_stream_release(request->request_stream);
             uint64_t offset = 0;
             size_t request_body_size = s_compute_request_body_size(meta_request, request->part_number, &offset);
             request->request_stream = aws_input_stream_new_from_parallel_stream(
-                allocator, meta_request->request_body_parallel_stream, offset, request_body_size);
+                allocator, meta_request->request_body_parallel_stream, meta_request, offset, request_body_size);
             s_s3_prepare_upload_part_finish(part_prep, AWS_ERROR_SUCCESS);
         }
     } else if (request->num_times_prepared == 0) {

source/s3_meta_request.c

@@ -319,7 +319,7 @@ int aws_s3_meta_request_init_base(
     } else if (options->send_using_async_writes == true) {
         meta_request->request_body_using_async_writes = true;
     }
-
+    aws_array_list_init_dynamic(&meta_request->read_metrics_list, allocator, 2048, sizeof(struct s3_data_read_metrics));
     meta_request->synced_data.next_streaming_part = 1;
 
     meta_request->meta_request_level_running_response_sum = NULL;

source/s3_parallel_input_stream.c

@@ -3,6 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0.
  */
 
+#include <aws/s3/private/s3_meta_request_impl.h>
 #include <aws/s3/private/s3_parallel_input_stream.h>
 
 #include <aws/common/atomics.h>
@@ -259,6 +260,10 @@ struct aws_s3_mmap_part_streaming_input_stream_impl {
 
     bool eos_loaded;
     bool eos_reached;
+
+    struct s3_data_read_metrics metrics;
+
+    struct aws_s3_meta_request *meta_request;
 };
 
 static int s_aws_s3_mmap_part_streaming_input_stream_seek(
@@ -276,6 +281,7 @@ static int s_aws_s3_mmap_part_streaming_input_stream_read(struct aws_input_strea
         AWS_CONTAINER_OF(stream, struct aws_s3_mmap_part_streaming_input_stream_impl, base);
     /* Map the content */
     size_t read_length = aws_min_size(dest->capacity - dest->len, impl->total_length - impl->total_length_read);
+    aws_high_res_clock_get_ticks(&impl->metrics.start_timestamp);
 
     if (impl->in_chunk_offset == SIZE_MAX) {
         /* The reading buf is invalid. Block until the loading buf is available. */
@@ -308,13 +314,22 @@ static int s_aws_s3_mmap_part_streaming_input_stream_read(struct aws_input_strea
         impl->in_chunk_offset = 0;
     }
     read_length = aws_min_size(read_length, impl->reading_chunk_buf->len - impl->in_chunk_offset);
+    impl->metrics.offset = impl->offset + impl->total_length_read;
+    impl->metrics.size = read_length;
+
     struct aws_byte_cursor chunk_cursor = aws_byte_cursor_from_buf(impl->reading_chunk_buf);
     aws_byte_cursor_advance(&chunk_cursor, impl->in_chunk_offset);
     chunk_cursor.len = read_length;
     aws_byte_buf_append(dest, &chunk_cursor);
     impl->in_chunk_offset += read_length;
     impl->total_length_read += read_length;
+    aws_high_res_clock_get_ticks(&impl->metrics.end_timestamp);
 
+    /* BEGIN CRITICAL SECTION */
+    aws_s3_meta_request_lock_synced_data(impl->meta_request);
+    aws_array_list_push_back(&impl->meta_request->read_metrics_list, &impl->metrics);
+    aws_s3_meta_request_unlock_synced_data(impl->meta_request);
+    /* END CRITICAL SECTION */
     if (impl->in_chunk_offset == impl->reading_chunk_buf->len) {
         /* We finished reading the reading buffer, reset it. */
         aws_byte_buf_reset(impl->reading_chunk_buf, false);
@@ -391,6 +406,7 @@ void aws_streaming_input_stream_reset(struct aws_input_stream *stream) {
 struct aws_input_stream *aws_input_stream_new_from_parallel_stream(
     struct aws_allocator *allocator,
     struct aws_parallel_input_stream *stream,
+    struct aws_s3_meta_request *meta_request,
     uint64_t offset,
     size_t request_body_size) {
 
@@ -412,6 +428,7 @@ struct aws_input_stream *aws_input_stream_new_from_parallel_stream(
     aws_byte_buf_init(&impl->chunk_buf_2, allocator, impl->chunk_load_size);
     impl->loading_chunk_buf = &impl->chunk_buf_1;
     impl->reading_chunk_buf = &impl->chunk_buf_2;
+    impl->meta_request = meta_request;
 
     /* Reset the input stream to start */
     aws_streaming_input_stream_reset(&impl->base);