Update to parameters; added window to reduce swaying.

Author: hoytak

cas_client/src/adaptive_concurrency_control.rs

@@ -43,10 +43,10 @@ impl ConcurrencyControllerState {
 
 /// A controller for robustly adjusting the amount of concurrancy on upload and download paths.
 ///
-/// By default, the controller dynamically adjusts the concurrency within bounds so that 80% of the
-/// transfers are completed within 20 seconds; it increases the concurrency as long as this criteria
-/// is met.  When more than 20% of the transfers begin taking longer than that, concurrency is reduced.
-/// Concurrency adjustments are throttled so that increasing the concurrency happens only every
+/// By default, the controller dynamically adjusts the concurrency within bounds so that between 70%
+/// and 90%  of the transfers are completed within 20 seconds; it increases the concurrency as long as
+/// this criteria is met.  When more than 20% of the transfers begin taking longer than that, concurrency
+/// is reduced. Concurrency adjustments are throttled so that increasing the concurrency happens only every
 /// 500ms, and decreasing it happens at most once every 250ms.   (These values are all defaults; see
 /// the constants and their definitions in constants.rs).
 ///  
@@ -169,7 +169,7 @@ impl AdaptiveConcurrencyController {
 
         let success_ratio = state_lg.add_report(success);
 
-        if success && (success_ratio > 0.8) {
+        if success && (success_ratio > *CONCURRENCY_CONTROL_TARGET_SUCCESS_RATIO_UPPER) {
             // Consider adjusting the concurrency
             if state_lg.last_adjustment_time.elapsed() > self.min_concurrency_increase_delay {
                 // Enough time has passed, so add a new permit.
@@ -183,22 +183,14 @@ impl AdaptiveConcurrencyController {
                     );
                 }
             }
-        } else {
+        } else if !success && (success_ratio < *CONCURRENCY_CONTROL_TARGET_SUCCESS_RATIO_LOWER) {
             // Had a failure, so attempt to decrease the number of permits.
             if state_lg.last_adjustment_time.elapsed() > self.min_concurrency_decrease_delay {
-                let decrease_reason = {
-                    if success {
-                        format!("success_ratio = {success_ratio:.2} < 0.8")
-                    } else {
-                        format!("failed transfer with success_ratio = {success_ratio:.2}")
-                    }
-                };
-
                 // Enough time has passed, so add a new permit.
                 if self.concurrency_semaphore.decrement_total_permits() {
                     state_lg.last_adjustment_time = Instant::now();
                     debug!(
-                        "Decreasing concurrency for {} to {} due to {decrease_reason}",
+                        "Decreasing concurrency for {} to {} due to failed transfer with success_ration = {success_ratio:.2}",
                         self.logging_tag,
                         self.concurrency_semaphore.total_permits()
                     );

cas_client/src/constants.rs

@@ -12,7 +12,7 @@ utils::configurable_constants! {
     ref CLIENT_RETRY_MAX_DURATION_MS: u64 = 6 * 60 * 1000; // 6m
 
     /// The target time for a small transfer to complete.
-    ref CONCURRENCY_CONTROL_TARGET_TIME_SMALL_TRANSFER_MS : u64 = 5 * 1000;
+    ref CONCURRENCY_CONTROL_TARGET_TIME_SMALL_TRANSFER_MS : u64 = 10 * 1000;
 
     /// The target time for a large transfer to complete.  Default is 20 seconds.
     ref CONCURRENCY_CONTROL_TARGET_TIME_LARGE_TRANSFER_MS : u64 = 20 * 1000;
@@ -27,7 +27,13 @@ utils::configurable_constants! {
     ref CONCURRENCY_CONTROL_MIN_DECREASE_WINDOW_MS : u64 = 250;
 
     /// The maximum number of connection successes and failures to examine when adjusting the concurrancy.
-    ref CONCURRENCY_CONTROL_TRACKING_SIZE : usize = 32;
+    ref CONCURRENCY_CONTROL_TRACKING_SIZE : usize = 20;
+
+    /// The maximum number of connection successes and failures to examine when adjusting the concurrancy.
+    ref CONCURRENCY_CONTROL_TARGET_SUCCESS_RATIO_LOWER: f64 = 0.7;
+
+    /// The maximum number of connection successes and failures to examine when adjusting the concurrancy.
+    ref CONCURRENCY_CONTROL_TARGET_SUCCESS_RATIO_UPPER: f64 = 0.9;
 
     /// The maximum time window within which to examine successes and failures when adjusting the concurrancy.
     ref CONCURRENCY_CONTROL_TRACKING_WINDOW_MS : u64 = 30 * 1000;

cas_client/src/remote_client.rs

@@ -618,6 +618,7 @@ impl Client for RemoteClient {
 
         let n_raw_bytes = serialized_cas_object.raw_num_bytes;
         let xorb_hash = serialized_cas_object.hash;
+        let n_transfer_bytes = serialized_cas_object.serialized_data.len() as u64;
 
         let progress_callback = move |bytes_sent: u64| {
             if let Some(utr) = upload_tracker.as_ref() {
@@ -641,7 +642,7 @@ impl Client for RemoteClient {
                 let api_tag = "cas::upload_xorb";
 
                 let response: UploadXorbResponse = RetryWrapper::new(api_tag)
-                    .with_connection_permit(upload_permit)
+                    .with_connection_permit(upload_permit, Some(n_transfer_bytes))
                     .run_and_extract_json(move || {
                         let upload_stream = upload_stream.clone_with_reset();
                         let url = url.clone();
@@ -806,7 +807,7 @@ impl Client for RemoteClient {
         let url = Url::parse(&format!("{}/shard/{key}", self.endpoint))?;
 
         let response: UploadShardResponse = RetryWrapper::new(api_tag)
-            .with_connection_permit(upload_permit)
+            .with_connection_permit(upload_permit, Some(shard_data.len() as u64))
             .run_and_extract_json(move || {
                 client
                     .request(method.clone(), url.clone())

cas_client/src/retry_wrapper.rs

@@ -26,7 +26,7 @@ pub struct RetryWrapper {
     no_retry_on_429: bool,
     log_errors_as_info: bool,
     api_tag: &'static str,
-    connection_permit: Option<Mutex<Option<ConnectionPermit>>>,
+    connection_permit: Option<Mutex<Option<(ConnectionPermit, Option<u64>)>>>,
 }
 
 impl RetryWrapper {
@@ -61,8 +61,8 @@ impl RetryWrapper {
         self
     }
 
-    pub fn with_connection_permit(mut self, permit: ConnectionPermit) -> Self {
-        self.connection_permit = Some(Mutex::new(Some(permit)));
+    pub fn with_connection_permit(mut self, permit: ConnectionPermit, n_bytes: Option<u64>) -> Self {
+        self.connection_permit = Some(Mutex::new(Some((permit, n_bytes))));
         self
     }
 
@@ -206,7 +206,7 @@ impl RetryWrapper {
 
                     if let Some(p) = &self_.connection_permit {
                         if let Some(p) = p.lock().await.as_mut() {
-                            p.transfer_starting()
+                            p.0.transfer_starting()
                         }
                     }
 
@@ -220,7 +220,7 @@ impl RetryWrapper {
                         Ok(resp) => self_.process_ok_response(try_idx, resp),
                     };
 
-                    let (n_bytes, processing_result) = match checked_result {
+                    let (reply_bytes, processing_result) = match checked_result {
                         Ok(ok_response) => (ok_response.content_length().unwrap_or(0), process_fn(ok_response).await),
                         Err(e) => (0, Err(e)),
                     };
@@ -231,17 +231,17 @@ impl RetryWrapper {
 
                         match &processing_result {
                             Ok(_) => {
-                                if let Some(permit) = maybe_permit.take() {
-                                    permit.report_completion(n_bytes, true).await;
+                                if let Some((permit, maybe_size)) = maybe_permit.take() {
+                                    permit.report_completion(maybe_size.unwrap_or(reply_bytes), true).await;
                                 }
                             },
                             Err(RetryableReqwestError::FatalError(_)) => {
-                                if let Some(permit) = maybe_permit.take() {
+                                if let Some((permit, _)) = maybe_permit.take() {
                                     permit.report_completion(0, false).await;
                                 }
                             },
                             Err(RetryableReqwestError::RetryableError(_)) => {
-                                if let Some(permit) = maybe_permit.as_ref() {
+                                if let Some((permit, _)) = maybe_permit.as_ref() {
                                     permit.report_retryable_failure().await;
                                 }
                             },