feat: add parent selector with connection management and capacity control

dragonfly-client/Cargo.toml

@@ -83,6 +83,8 @@ http-body-util = "0.1.3"
 termion = "4.0.5"
 tabled = "0.19.0"
 path-absolutize = "3.1.1"
+dashmap = "6.1.0"
+rand = "0.9.1"
 
 [dev-dependencies]
 tempfile.workspace = true

dragonfly-client/src/bin/dfdaemon/main.rs

@@ -25,7 +25,9 @@ use dragonfly_client::grpc::{
 use dragonfly_client::health::Health;
 use dragonfly_client::metrics::Metrics;
 use dragonfly_client::proxy::Proxy;
-use dragonfly_client::resource::{persistent_cache_task::PersistentCacheTask, task::Task};
+use dragonfly_client::resource::{
+    parent_selector::ParentSelector, persistent_cache_task::PersistentCacheTask, task::Task,
+};
 use dragonfly_client::shutdown;
 use dragonfly_client::stats::Stats;
 use dragonfly_client::tracing::init_tracing;
@@ -209,13 +211,26 @@ async fn main() -> Result<(), anyhow::Error> {
         })?;
     let backend_factory = Arc::new(backend_factory);
 
+    // Initialize parent selector.
+    let parent_selector = ParentSelector::new(
+        config.clone(),
+        id_generator.clone(),
+        shutdown.clone(),
+        shutdown_complete_tx.clone(),
+    )
+    .inspect_err(|err| {
+        error!("initialize parent selector failed: {}", err);
+    })?;
+    let parent_selector = Arc::new(parent_selector);
+
     // Initialize task manager.
     let task = Task::new(
         config.clone(),
         id_generator.clone(),
         storage.clone(),
         scheduler_client.clone(),
         backend_factory.clone(),
+        parent_selector.clone(),
     )?;
     let task = Arc::new(task);
 
@@ -372,6 +387,15 @@ async fn main() -> Result<(), anyhow::Error> {
             info!("proxy server exited");
         },
 
+        _ = {
+            let barrier = grpc_server_started_barrier.clone();
+            tokio::spawn(async move {
+                parent_selector.run(barrier).await.unwrap_or_else(|err| error!("parent selector failed: {}", err));
+            })
+        } => {
+            info!("parent selector exited");
+        },
+
         _ = shutdown::shutdown_signal() => {},
     }
 

dragonfly-client/src/grpc/dfdaemon_upload.rs

@@ -984,12 +984,14 @@ impl DfdaemonUpload for DfdaemonUploadServerHandler {
                     let mut host = Host::default();
                     if let Some(network_data) = networks.get(&interface.name) {
                         let network = Network {
-                            download_rate: network_data.received()
-                                / DEFAULT_HOST_INFO_REFRESH_INTERVAL.as_secs(),
+                            download_rate: (network_data.received() as f64
+                                / DEFAULT_HOST_INFO_REFRESH_INTERVAL.as_secs_f64())
+                                as u64,
                             // Convert bandwidth to bytes per second.
                             download_rate_limit: interface.bandwidth / 8 * MB,
-                            upload_rate: network_data.transmitted()
-                                / DEFAULT_HOST_INFO_REFRESH_INTERVAL.as_secs(),
+                            upload_rate: (network_data.transmitted() as f64
+                                / DEFAULT_HOST_INFO_REFRESH_INTERVAL.as_secs_f64())
+                                as u64,
                             // Convert bandwidth to bytes per second.
                             upload_rate_limit: interface.bandwidth / 8 * MB,
                             ..Default::default()

dragonfly-client/src/resource/mod.rs

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 
+pub mod parent_selector;
 pub mod persistent_cache_task;
 pub mod piece;
 pub mod piece_collector;