feat: Add CacheTask processing logic

## Description
This pull request adds the CacheTask task type, completes the adaptation for CacheTask during gRPC download and piece processing, and finally adds a cache parameter to dfget for creating and managing CacheTask.

## Related Issue

### Changes
- Added --cache parameter to dfget to create and manage CacheTask.
- Added cache task and implemented its download logic.
Implemented piece download execution for cache tasks in piece.rs and piece_download.rs.
- Added handling for cache peer and cache task in dragonfly-client/grpc/scheduler.rs.
- Added cache task support to dfdaemon gRPC servers.

## Motivation and Context
To introduce the CacheTask task type to avoid disk interactions, thereby increasing response speed.

Signed-off-by: fu220 <[email protected]>

Author: fu220

dragonfly-client-util/src/id_generator/mod.rs

@@ -31,6 +31,9 @@ const SEED_PEER_SUFFIX: &str = "seed";
 /// PERSISTENT_CACHE_TASK_SUFFIX is the suffix of the persistent cache task.
 const PERSISTENT_CACHE_TASK_SUFFIX: &str = "persistent-cache-task";
 
+/// CACHE_TASK_SUFFIX is the suffix of the cache task.
+const CACHE_TASK_SUFFIX: &str = "cache-task";
+
 /// TaskIDParameter is the parameter of the task id.
 pub enum TaskIDParameter {
     /// Content uses the content to generate the task id.
@@ -59,6 +62,21 @@ pub enum PersistentCacheTaskIDParameter {
     },
 }
 
+/// CacheTaskIDParameter is the parameter of the cache task id.
+pub enum CacheTaskIDParameter {
+    /// Content uses the content to generate the cache task id.
+    Content(String),
+    /// URLBased uses the url, piece_length, tag, application and filtered_query_params to generate
+    /// the cache task id.
+    URLBased {
+        url: String,
+        piece_length: Option<u64>,
+        tag: Option<String>,
+        application: Option<String>,
+        filtered_query_params: Vec<String>,
+    },
+}
+
 /// IDGenerator is used to generate the id for the resources.
 #[derive(Debug)]
 pub struct IDGenerator {
@@ -210,6 +228,69 @@ impl IDGenerator {
         }
     }
 
+    /// cache_task_id generates the cache task id.
+    #[inline]
+    pub fn cache_task_id(&self, parameter: CacheTaskIDParameter) -> Result<String> {
+        match parameter {
+            CacheTaskIDParameter::Content(content) => {
+                Ok(hex::encode(Sha256::digest(content.as_bytes())))
+            }
+            CacheTaskIDParameter::URLBased {
+                url,
+                piece_length,
+                tag,
+                application,
+                filtered_query_params,
+            } => {
+                // Filter the query parameters.
+                let url = Url::parse(url.as_str()).or_err(ErrorType::ParseError)?;
+                let query = url
+                    .query_pairs()
+                    .filter(|(k, _)| !filtered_query_params.contains(&k.to_string()));
+
+                let mut artifact_url = url.clone();
+                if query.clone().count() == 0 {
+                    artifact_url.set_query(None);
+                } else {
+                    artifact_url.query_pairs_mut().clear().extend_pairs(query);
+                }
+
+                let artifact_url_str = artifact_url.to_string();
+                let final_url = if artifact_url_str.ends_with('/') && artifact_url.path() == "/" {
+                    artifact_url_str.trim_end_matches('/').to_string()
+                } else {
+                    artifact_url_str
+                };
+
+                // Initialize the hasher.
+                let mut hasher = Sha256::new();
+
+                // Add the url to generate the cache task id.
+                hasher.update(final_url);
+
+                // Add the tag to generate the cache task id.
+                if let Some(tag) = tag {
+                    hasher.update(tag);
+                }
+
+                // Add the application to generate the cache task id.
+                if let Some(application) = application {
+                    hasher.update(application);
+                }
+
+                // Add the piece length to generate the cache task id.
+                if let Some(piece_length) = piece_length {
+                    hasher.update(piece_length.to_string());
+                }
+
+                hasher.update(TaskType::Cache.as_str_name().as_bytes());
+
+                // Generate the cache task id.
+                Ok(hex::encode(hasher.finalize()))
+            }
+        }
+    }
+
     /// peer_id generates the peer id.
     #[inline]
     pub fn peer_id(&self) -> String {
@@ -232,6 +313,10 @@ impl IDGenerator {
             return TaskType::PersistentCache;
         }
 
+        if id.ends_with(CACHE_TASK_SUFFIX) {
+            return TaskType::Cache;
+        }
+
         TaskType::Standard
     }
 }
@@ -434,6 +519,7 @@ mod tests {
                 "some-task-id-persistent-cache-task",
                 TaskType::PersistentCache,
             ),
+            ("some-task-id-cache-task", TaskType::Cache),
         ];
 
         let generator = IDGenerator::new("127.0.0.1".to_string(), "localhost".to_string(), false);

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::{
+    cache_task::CacheTask, persistent_cache_task::PersistentCacheTask, task::Task,
+};
 use dragonfly_client::stats::Stats;
 use dragonfly_client::tracing::init_tracing;
 use dragonfly_client_backend::BackendFactory;
@@ -218,6 +220,16 @@ async fn main() -> Result<(), anyhow::Error> {
     )?;
     let task = Arc::new(task);
 
+    // Initialize cache task manager.
+    let cache_task = CacheTask::new(
+        config.clone(),
+        id_generator.clone(),
+        storage.clone(),
+        scheduler_client.clone(),
+        backend_factory.clone(),
+    )?;
+    let cache_task = Arc::new(cache_task);
+
     // Initialize persistent cache task manager.
     let persistent_cache_task = PersistentCacheTask::new(
         config.clone(),
@@ -279,6 +291,7 @@ async fn main() -> Result<(), anyhow::Error> {
         config.clone(),
         SocketAddr::new(config.upload.server.ip.unwrap(), config.upload.server.port),
         task.clone(),
+        cache_task.clone(),
         persistent_cache_task.clone(),
         interface.clone(),
         shutdown.clone(),
@@ -290,6 +303,7 @@ async fn main() -> Result<(), anyhow::Error> {
         config.clone(),
         config.download.server.socket_path.clone(),
         task.clone(),
+        cache_task.clone(),
         persistent_cache_task.clone(),
         shutdown.clone(),
         shutdown_complete_tx.clone(),