Added comments and corrected typos.

Author: hoytak

cas_client/src/adaptive_concurrency/controller.rs

@@ -44,7 +44,7 @@ impl ConcurrencyControllerState {
     }
 }
 
-/// A controller for dynamically adjusting the amount of concurrancy on upload and download paths.
+/// A controller for dynamically adjusting the amount of concurrency on upload and download paths.
 ///
 /// This controller uses two statistical models that adapt over time using exponentially weighted
 /// moving averages.  The first is a model that predicts the overall current bandwith, and the second is
@@ -99,7 +99,7 @@ impl AdaptiveConcurrencyController {
         })
     }
 
-    /// Acquire a connection permit based on the current concurrancy.
+    /// Acquire a connection permit based on the current concurrency.
     pub async fn acquire_connection_permit(self: &Arc<Self>) -> Result<ConnectionPermit, CasClientError> {
         let permit = self.concurrency_semaphore.acquire().await?;
 

cas_client/src/constants.rs

@@ -45,7 +45,7 @@ utils::configurable_constants! {
     ref MAX_CONCURRENT_UPLOADS: usize = 100;
 
     /// The minimum number of simultaneous xorb and/or shard upload streams that the
-    /// the adaptive concurrency control may reduce the concurrancy down to on slower connections.
+    /// the adaptive concurrency control may reduce the concurrency down to on slower connections.
     ref MIN_CONCURRENT_UPLOADS: usize = 2;
 
     /// The starting number of concurrent upload streams, which will increase up to MAX_CONCURRENT_UPLOADS

cas_client/src/retry_wrapper.rs

@@ -228,6 +228,7 @@ impl RetryWrapper {
                         Ok(resp) => self_.process_ok_response(try_idx, resp),
                     };
 
+                    // reply_bytes is ignored if the size was specified earlier, as in the case for upload.
                     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)),

data/src/file_upload_session.rs

@@ -184,7 +184,7 @@ impl FileUploadSession {
             let file_id = self.completion_tracker.register_new_file(file_name.clone(), file_size).await;
 
             // Now, spawn a task
-            let ingestion_concurrancy_limiter = file_parallel_limiter.clone();
+            let ingestion_concurrency_limiter = file_parallel_limiter.clone();
             let session = self.clone();
 
             cleaning_tasks.push(tokio::spawn(async move {
@@ -201,7 +201,7 @@ impl FileUploadSession {
                     "file.defrag_prevented_dedup_chunks" = tracing::field::Empty,
                 );
                 // First, get a permit to process this file.
-                let _processing_permit = ingestion_concurrancy_limiter.acquire().await?;
+                let _processing_permit = ingestion_concurrency_limiter.acquire().await?;
 
                 async move {
                     let mut reader = File::open(&file_path)?;

utils/src/adjustable_semaphore.rs

@@ -5,9 +5,11 @@ use std::sync::Arc;
 use tokio::sync::{AcquireError, Semaphore};
 
 /// An adjustable semaphore in which the total number of permits can be adjusted at any time
-/// between a minimum and a maximum bound.  Adjustments do not affect any permits currently
-/// issued; if an adjustment cannot be permformed immediately, then it is resolved before any
-/// new permits are issued.
+/// between a minimum and a maximum bound.  
+///
+/// Unlike the tokio Semaphore, decreasing the number of permits may be done at any time and
+/// are resolved lazily if needed; any permits currently issued remain valid, but no new permits
+/// are issued until any requested decreases are resolved.
 pub struct AdjustableSemaphore {
     semaphore: Arc<Semaphore>,
     total_permits: AtomicUsize,
@@ -17,7 +19,7 @@ pub struct AdjustableSemaphore {
 }
 
 /// A permit issued by the AdjustableSemaphore.  On drop, this attempts to
-/// resolve an enqueued permit decrease if one is needed.
+/// resolve any enqueued permit decrease if one is needed.
 pub struct AdjustableSemaphorePermit {
     permit: Option<tokio::sync::OwnedSemaphorePermit>,
     parent: Arc<AdjustableSemaphore>,
@@ -65,6 +67,11 @@ impl AdjustableSemaphore {
         // A few debug mode consistency checks.
         debug_assert!(self.semaphore.available_permits() <= self.max_permits);
 
+        // To ensure that the fairness property of the enclosed tokio semaphore is respected,
+        // this function must only call this class and not attempt to resolve anything else.
+        //
+        // As a result, any decreases are resolved by the Drop trait of this permit, which may
+        // mean that permit is forgotten to resolve an outstanding decrease.
         let permit = self.semaphore.clone().acquire_owned().await?;
 
         Ok(AdjustableSemaphorePermit {