use locks instead of atomic var

Author: Krish

include/aws/s3/private/s3_client_impl.h

@@ -376,7 +376,8 @@ struct aws_s3_client {
         struct aws_atomic_var num_requests_streaming_request_body;
 
         /* Total weight of all active meta requests. Weight = object_size / (part_size * part_size) */
-        struct aws_atomic_var total_weight;
+        double total_weight;
+        struct aws_mutex total_weight_mutex;
 
         /* Sum of connection requirements from all active meta requests */
         struct aws_atomic_var total_required_connections;

source/s3_auto_ranged_get.c

@@ -118,6 +118,16 @@ struct aws_s3_meta_request *aws_s3_meta_request_auto_ranged_get_new(
     if (options->object_size_hint != NULL) {
         auto_ranged_get->object_size_hint_available = true;
         auto_ranged_get->object_size_hint = *options->object_size_hint;
+
+        /* Calculate weight early if object size hint is available: weight = object_size / (part_size * part_size) */
+        if (auto_ranged_get->object_size_hint > 0 && auto_ranged_get->base.part_size > 0) {
+            auto_ranged_get->base.weight =
+                (double)auto_ranged_get->object_size_hint /
+                ((double)auto_ranged_get->base.part_size * (double)auto_ranged_get->base.part_size);
+            aws_mutex_lock(&client->stats.total_weight_mutex);
+            client->stats.total_weight += auto_ranged_get->base.weight;
+            aws_mutex_unlock(&client->stats.total_weight_mutex);
+        }
     }
     AWS_LOGF_DEBUG(
         AWS_LS_S3_META_REQUEST, "id=%p Created new Auto-Ranged Get Meta Request.", (void *)&auto_ranged_get->base);
@@ -969,11 +979,14 @@ static void s_s3_auto_ranged_get_request_finished(
             }
 
             /* Calculate weight for download: weight = object_size / (part_size * part_size) */
-            if (object_size > 0 && meta_request->part_size > 0) {
+            /* Only calculate if weight wasn't already set from object_size_hint */
+            if (object_size > 0 && meta_request->part_size > 0 && meta_request->weight == 0.0) {
                 meta_request->weight =
                     (double)object_size / ((double)meta_request->part_size * (double)meta_request->part_size);
                 /* Add this meta request's weight to client's total weight */
-                aws_atomic_fetch_add(&meta_request->client->stats.total_weight, (size_t)meta_request->weight);
+                aws_mutex_lock(&meta_request->client->stats.total_weight_mutex);
+                meta_request->client->stats.total_weight += meta_request->weight;
+                aws_mutex_unlock(&meta_request->client->stats.total_weight_mutex);
             }
         }
 

source/s3_auto_ranged_put.c

@@ -371,7 +371,9 @@ struct aws_s3_meta_request *aws_s3_meta_request_auto_ranged_put_new(
     if (has_content_length && part_size > 0) {
         auto_ranged_put->base.weight = (double)content_length / ((double)part_size * (double)part_size);
         /* Add this meta request's weight to client's total weight */
-        aws_atomic_fetch_add(&client->stats.total_weight, (size_t)auto_ranged_put->base.weight);
+        aws_mutex_lock(&client->stats.total_weight_mutex);
+        client->stats.total_weight += auto_ranged_put->base.weight;
+        aws_mutex_unlock(&client->stats.total_weight_mutex);
     } else {
         auto_ranged_put->base.weight = 0.0;
     }

source/s3_client.c

@@ -628,7 +628,8 @@ struct aws_s3_client *aws_s3_client_new(
 
     aws_atomic_init_int(&client->stats.num_requests_stream_queued_waiting, 0);
     aws_atomic_init_int(&client->stats.num_requests_streaming_response, 0);
-    aws_atomic_init_int(&client->stats.total_weight, 0);
+    client->stats.total_weight = 0.0;
+    aws_mutex_init(&client->stats.total_weight_mutex);
     aws_atomic_init_int(&client->stats.total_required_connections, 0);
     aws_atomic_init_int(&client->stats.cached_max_connections, 0);
     aws_atomic_init_int(&client->stats.last_max_connections_update_ns, 0);
@@ -862,6 +863,7 @@ struct aws_s3_client *aws_s3_client_new(
     }
     aws_client_bootstrap_release(client->client_bootstrap);
     aws_mutex_clean_up(&client->synced_data.lock);
+    aws_mutex_clean_up(&client->stats.total_weight_mutex);
 
     aws_mem_release(client->allocator, client->network_interface_names_cursor_array);
     for (size_t i = 0; i < aws_array_list_length(&client->network_interface_names); i++) {
@@ -966,6 +968,7 @@ static void s_s3_client_finish_destroy_default(struct aws_s3_client *client) {
     aws_mem_release(client->allocator, client->tcp_keep_alive_options);
 
     aws_mutex_clean_up(&client->synced_data.lock);
+    aws_mutex_clean_up(&client->stats.total_weight_mutex);
 
     AWS_ASSERT(aws_linked_list_empty(&client->synced_data.pending_meta_request_work));
     AWS_ASSERT(aws_linked_list_empty(&client->threaded_data.meta_requests));
@@ -2369,6 +2372,7 @@ void aws_s3_client_update_connections_threaded(struct aws_s3_client *client) {
 
         struct aws_s3_request *request = aws_s3_client_dequeue_request_threaded(client);
         struct aws_s3_meta_request *meta_request = request->meta_request;
+        const uint32_t max_active_connections = aws_s3_client_get_max_active_connections(client, meta_request);
         /* As the request removed from the queue. Decrement the preparing track */
         --meta_request->client_process_work_threaded_data.num_request_being_prepared;
         if (request->is_noop) {
@@ -2395,18 +2399,16 @@ void aws_s3_client_update_connections_threaded(struct aws_s3_client *client) {
             bool should_allocate_connection = false;
             uint32_t current_connections = (uint32_t)aws_atomic_load_int(&meta_request->num_requests_network);
 
-            if (meta_request->weight > 0.0) {
-                size_t total_weight = aws_atomic_load_int(&client->stats.total_weight);
-                if (total_weight > 0) {
+            if (meta_request->weight > (double)0) {
+                aws_mutex_lock(&client->stats.total_weight_mutex);
+                double total_weight = client->stats.total_weight;
+                aws_mutex_unlock(&client->stats.total_weight_mutex);
+                if (total_weight > (double)0) {
                     /* Calculate: (meta_request->weight / client->total_weight) * 300 */
-                    double weight_ratio = meta_request->weight / (double)total_weight;
+                    double weight_ratio = meta_request->weight / total_weight;
                     uint32_t allowed_connections = (uint32_t)(weight_ratio * client_max_active_connections);
 
-                    /* Apply max_active_connections_override if set */
-                    if (meta_request->max_active_connections_override > 0) {
-                        allowed_connections =
-                            aws_min_u32(allowed_connections, meta_request->max_active_connections_override);
-                    }
+                    allowed_connections = aws_min_u32(allowed_connections, max_active_connections);
 
                     should_allocate_connection = (current_connections < allowed_connections);
                 } else {

source/s3_meta_request.c

@@ -615,7 +615,9 @@ static void s_s3_meta_request_destroy(void *user_data) {
     if (meta_request->client != NULL) {
         /* Subtract this meta request's weight from client's total weight */
         if (meta_request->weight > 0.0) {
-            aws_atomic_fetch_sub(&meta_request->client->stats.total_weight, (size_t)meta_request->weight);
+            aws_mutex_lock(&meta_request->client->stats.total_weight_mutex);
+            meta_request->client->stats.total_weight -= meta_request->weight;
+            aws_mutex_unlock(&meta_request->client->stats.total_weight_mutex);
         }
 
         /* Subtract this meta request's connection requirements from client's total */