bootstrap new scheduler (#131)

bootstrap new scheduler

Add a scheduler module that coordinates transfer execution: work
generation, prioritization, capacity gating, and submission. This sits
alongside the existing runtime::scheduler and is not yet wired into any
operations.

Transfers implement a state machine trait (poll_work/execute). The
scheduler polls them for work when capacity is available. Work is
generated lazily, one item per poll, so memory and in-flight work are
naturally bounded without the scheduler knowing operation internals.

Ordering uses CFS (Completely Fair Scheduling) adapted from the Linux
kernel. Each transfer accumulates virtual runtime weighted by priority.
The ready set is a skiplist ordered by vruntime; the scheduler always
picks the transfer with the least scheduling share. Priority-1 transfers
still make progress, they just accumulate vruntime faster.

The ready set is edge-triggered: transfers enter on enqueue or wake,
leave on Pending or Done. Scheduling cost scales with active transfers,
not total count.

Capacity is gated by a ConcurrencyController trait. FixedConcurrency
returns a constant target. The trait is the seam for adaptive concurrency
(not in this change).

Follow-on work (e.g. disk read completes, now send over network) bypasses
the ready set and goes directly to execution. CFS controls admission;
successors complete admitted work without re-competing.

Benchmarked on c6in.16xlarge: matches main branch throughput (55 Gb/s)
at 60% less memory (4.6 GB vs 11.8 GB peak RSS) with seq window
backpressure bounding out-of-order buffering.

Design: docs/design/scheduler.md

Author: aajtodd

Cargo.lock

@@ -26,6 +26,7 @@ aws-types = "1.3.7"
 blocking = "1.6.1"
 bytes = "1"
 bytes-utils = "0.1.4"
+crossbeam-skiplist = "0.1"
 futures-util = "0.3.31"
 path-clean = "1.0.1"
 pin-project-lite = "0.2.16"

aws-sdk-s3-transfer-manager/Cargo.toml

@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-use crate::runtime::scheduler::Scheduler;
+use crate::runtime::scheduler::Scheduler as LegacyScheduler;
 use crate::types::{ConcurrencyMode, PartSize};
 use crate::Config;
 use crate::{metrics::unit::ByteUnit, DEFAULT_CONCURRENCY};
@@ -19,7 +19,8 @@ pub struct Client {
 #[derive(Debug)]
 pub(crate) struct Handle {
     pub(crate) config: crate::Config,
-    pub(crate) scheduler: Scheduler,
+    pub(crate) scheduler: crate::scheduler::Scheduler,
+    pub(crate) legacy_scheduler: LegacyScheduler,
 }
 
 impl Handle {
@@ -63,8 +64,19 @@ impl Handle {
 impl Client {
     /// Creates a new client from a transfer manager config.
     pub fn new(config: Config) -> Client {
-        let scheduler = Scheduler::new(config.concurrency().clone());
-        let handle = Arc::new(Handle { config, scheduler });
+        use crate::scheduler::FixedConcurrency;
+        let new_scheduler = crate::scheduler::Scheduler::with_controller(Arc::new(
+            FixedConcurrency::new(match config.concurrency() {
+                ConcurrencyMode::Explicit(c) => *c,
+                _ => crate::DEFAULT_CONCURRENCY,
+            }),
+        ));
+        let legacy_scheduler = LegacyScheduler::new(config.concurrency().clone());
+        let handle = Arc::new(Handle {
+            config,
+            scheduler: new_scheduler,
+            legacy_scheduler,
+        });
         Client { handle }
     }
 

aws-sdk-s3-transfer-manager/src/client.rs

@@ -92,6 +92,9 @@ pub(crate) mod http;
 /// Internal runtime components
 pub(crate) mod runtime;
 
+/// Scheduler for coordinating transfer execution
+pub(crate) mod scheduler;
+
 /// Metrics
 pub mod metrics;
 

aws-sdk-s3-transfer-manager/src/lib.rs

@@ -127,7 +127,7 @@ async fn send_discovery(
     // acquire a permit for discovery
     let permit = ctx
         .handle
-        .scheduler
+        .legacy_scheduler
         .acquire_permit(PermitType::Network(NetworkPermitContext {
             payload_size_estimate: 0,
             bucket_type: ctx.bucket_type(),

aws-sdk-s3-transfer-manager/src/operation/download.rs

@@ -154,7 +154,7 @@ pub(super) fn chunk_service(
        + Clone
        + Send {
     let svc = service_fn(download_chunk_handler);
-    let concurrency_limit = ConcurrencyLimitLayer::new(ctx.handle.scheduler.clone());
+    let concurrency_limit = ConcurrencyLimitLayer::new(ctx.handle.legacy_scheduler.clone());
 
     ServiceBuilder::new()
         .layer(concurrency_limit)

aws-sdk-s3-transfer-manager/src/operation/download/service.rs

@@ -117,7 +117,7 @@ async fn put_object(
 
     let _permit = ctx
         .handle
-        .scheduler
+        .legacy_scheduler
         .acquire_permit(PermitType::Network(NetworkPermitContext {
             payload_size_estimate: content_length as u64,
             bucket_type: ctx.state.bucket_type(),

aws-sdk-s3-transfer-manager/src/operation/upload.rs

@@ -109,7 +109,7 @@ pub(super) fn upload_part_service(
        + Clone
        + Send {
     let svc = service_fn(upload_part_handler);
-    let concurrency_limit = ConcurrencyLimitLayer::new(ctx.handle.scheduler.clone());
+    let concurrency_limit = ConcurrencyLimitLayer::new(ctx.handle.legacy_scheduler.clone());
 
     let svc = ServiceBuilder::new()
         .layer(concurrency_limit)
@@ -239,7 +239,12 @@ mod tests {
             ctx: UploadContext {
                 handle: Arc::new(Handle {
                     config: Config::builder().client(s3_client).build(),
-                    scheduler: Scheduler::new(ConcurrencyMode::Explicit(1)),
+                    scheduler: crate::scheduler::Scheduler::with_controller(Arc::new(
+                        crate::scheduler::FixedConcurrency::new(1),
+                    )),
+                    legacy_scheduler: crate::runtime::scheduler::Scheduler::new(
+                        ConcurrencyMode::Explicit(1),
+                    ),
                 }),
                 state: Arc::new(UploadState {
                     request: Arc::new(

aws-sdk-s3-transfer-manager/src/operation/upload/service.rs

@@ -328,6 +328,7 @@ mod tests {
     use aws_sdk_s3::operation::put_object::PutObjectOutput;
     use aws_smithy_mocks::{mock, mock_client, RuleMode};
     use bytes::Bytes;
+    use std::sync::Arc;
 
     use crate::{
         client::Handle,
@@ -711,9 +712,18 @@ mod tests {
         let s3_client = mock_client!(aws_sdk_s3, RuleMode::MatchAny, &[put_object]);
         let config = crate::Config::builder().client(s3_client).build();
 
-        let scheduler = Scheduler::new(ConcurrencyMode::Explicit(DEFAULT_CONCURRENCY));
-
-        let handle = std::sync::Arc::new(Handle { config, scheduler });
+        let legacy_scheduler = crate::runtime::scheduler::Scheduler::new(
+            ConcurrencyMode::Explicit(DEFAULT_CONCURRENCY),
+        );
+        let scheduler = crate::scheduler::Scheduler::with_controller(Arc::new(
+            crate::scheduler::FixedConcurrency::new(DEFAULT_CONCURRENCY),
+        ));
+
+        let handle = std::sync::Arc::new(Handle {
+            config,
+            scheduler,
+            legacy_scheduler,
+        });
         let input = UploadObjectsInputBuilder::default()
             .source("doesnotmatter")
             .bucket(bucket)

aws-sdk-s3-transfer-manager/src/operation/upload_objects/worker.rs

@@ -0,0 +1,36 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+use std::fmt;
+
+/// Controls how many work items can be in-flight concurrently.
+///
+/// The scheduler calls `target()` before generating work. Work is only
+/// generated when total in-flight + pending is below the target.
+///
+/// `FixedConcurrency` provides a constant target. Future implementations
+/// (e.g., `AdaptiveConcurrency`) can adjust the target dynamically based
+/// on observed throughput.
+pub(crate) trait ConcurrencyController: Send + Sync + fmt::Debug {
+    /// Current concurrency target. May change between calls.
+    fn target(&self) -> usize;
+}
+
+/// Fixed concurrency target that never changes.
+#[derive(Debug)]
+pub(crate) struct FixedConcurrency(usize);
+
+impl FixedConcurrency {
+    pub(crate) fn new(target: usize) -> Self {
+        assert!(target > 0, "concurrency target must be at least 1");
+        Self(target)
+    }
+}
+
+impl ConcurrencyController for FixedConcurrency {
+    fn target(&self) -> usize {
+        self.0
+    }
+}