Replace gRPC download call in proxy with local download_task using task manager

dragonfly-client-core/src/error/mod.rs

@@ -158,6 +158,10 @@ pub enum DFError {
     #[error{"unimplemented"}]
     Unimplemented,
 
+    /// The error when the permission is denied.
+    #[error{"permission denied"}]
+    PermissionDenied,
+
     /// The error when the range fallback error is empty.
     #[error{"RangeUnsatisfiable: Failed to parse range fallback error, please file an issue"}]
     EmptyHTTPRangeError,

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

@@ -368,6 +368,7 @@ async fn main() -> Result<(), anyhow::Error> {
     let proxy = Proxy::new(
         config.clone(),
         task.clone(),
+        dynconfig.clone(),
         shutdown.clone(),
         shutdown_complete_tx.clone(),
     );

dragonfly-client/src/grpc/block_list.rs → dragonfly-client/src/dynconfig/block_list.rs

@@ -14,12 +14,13 @@
  * limitations under the License.
  */
 
-use crate::dynconfig::{
-    Dynconfig, SchedulerClusterConfigBlockList, SchedulerClusterConfigDownloadBlockList,
+use super::{
+    Data, SchedulerClusterConfigBlockList, SchedulerClusterConfigDownloadBlockList,
     SchedulerClusterConfigUploadBlockList,
 };
 use dragonfly_client_config::dfdaemon::Config;
 use std::sync::Arc;
+use tokio::sync::RwLock;
 
 /// Parameters for checking download block list.
 #[derive(Debug, Clone)]
@@ -55,16 +56,17 @@ pub struct BlockList {
     /// Configuration of the dfdaemon.
     config: Arc<Config>,
 
-    /// Dynamic configuration of the dfdaemon.
-    dynconfig: Arc<Dynconfig>,
+    /// Dynamic configuration data shared with the dynconfig.
+    data: Arc<RwLock<Data>>,
 }
 
 /// The block list struct provides methods to check if certain tasks are blocked based on the
 /// dynamic configuration.
 impl BlockList {
-    /// Creates a new `BlockList` instance with the given static configuration and dynamic configuration.
-    pub fn new(config: Arc<Config>, dynconfig: Arc<Dynconfig>) -> Self {
-        Self { config, dynconfig }
+    /// Creates a new `BlockList` instance with the given static configuration and shared dynamic
+    /// configuration data.
+    pub fn new(config: Arc<Config>, data: Arc<RwLock<Data>>) -> Self {
+        Self { config, data }
     }
 
     /// Acquires a read lock on the dynamic configuration data and applies the given transformation
@@ -75,7 +77,7 @@ impl BlockList {
     where
         F: FnOnce(&SchedulerClusterConfigBlockList) -> Option<R>,
     {
-        let data = self.dynconfig.data.read().await;
+        let data = self.data.read().await;
         let config = if self.config.seed_peer.enable {
             data.seed_client_config
                 .as_ref()

dragonfly-client/src/dynconfig/mod.rs

@@ -16,6 +16,7 @@
 
 use crate::grpc::health::HealthClient;
 use crate::grpc::manager::ManagerClient;
+use block_list::BlockList;
 use dragonfly_api::manager::v2::{
     ListSchedulersRequest, ListSchedulersResponse, Scheduler, SourceType,
 };
@@ -33,6 +34,8 @@ use tonic_health::pb::health_check_response::ServingStatus;
 use tracing::{debug, error, info, instrument};
 use url::Url;
 
+pub mod block_list;
+
 /// Block list configuration for scheduler cluster clients.
 #[derive(Debug, Clone, Default, Serialize, Deserialize)]
 #[serde(default)]
@@ -149,7 +152,10 @@ pub struct Data {
 /// refreshing state from the manager service.
 pub struct Dynconfig {
     /// Current dynamic configuration, protected by a read-write lock.
-    pub data: RwLock<Data>,
+    pub data: Arc<RwLock<Data>>,
+
+    /// Block list to check whether tasks are blocked, backed by the dynamic configuration data.
+    pub block_list: Arc<BlockList>,
 
     /// Static dfdaemon configuration.
     config: Arc<Config>,
@@ -177,9 +183,11 @@ impl Dynconfig {
         shutdown_complete_tx: mpsc::UnboundedSender<()>,
     ) -> Result<Self> {
         // Create a new Dynconfig.
+        let data = Arc::new(RwLock::new(Data::default()));
         let dc = Dynconfig {
+            block_list: Arc::new(BlockList::new(config.clone(), data.clone())),
+            data,
             config,
-            data: RwLock::new(Data::default()),
             manager_client,
             mutex: Mutex::new(()),
             shutdown,

dragonfly-client/src/grpc/dfdaemon_download.rs

@@ -14,10 +14,8 @@
  * limitations under the License.
  */
 
+use crate::dynconfig::block_list::{DownloadBlockListCheckParams, UploadBlockListCheckParams};
 use crate::dynconfig::Dynconfig;
-use crate::grpc::block_list::{
-    BlockList, DownloadBlockListCheckParams, UploadBlockListCheckParams,
-};
 use crate::resource::{persistent_cache_task, persistent_task, task};
 use dragonfly_api::common::v2::{
     CacheTask, PersistentCacheTask, PersistentTask, Priority, Task, TaskType,
@@ -165,7 +163,7 @@ impl DfdaemonDownloadServer {
         let service = DfdaemonDownloadGRPCServer::with_interceptor(
             DfdaemonDownloadServerHandler {
                 config: self.config.clone(),
-                block_list: BlockList::new(self.config.clone(), self.dynconfig.clone()),
+                dynconfig: self.dynconfig.clone(),
                 socket_path: self.socket_path.clone(),
                 task: self.task.clone(),
                 persistent_task: self.persistent_task.clone(),
@@ -273,8 +271,8 @@ pub struct DfdaemonDownloadServerHandler {
     /// Configuration of the dfdaemon.
     config: Arc<Config>,
 
-    /// Block list for download tasks.
-    block_list: BlockList,
+    /// Dynamic configuration of the dfdaemon.
+    dynconfig: Arc<Dynconfig>,
 
     /// Path of the Unix domain socket.
     socket_path: PathBuf,
@@ -329,6 +327,7 @@ impl DfdaemonDownload for DfdaemonDownloadServerHandler {
             priority: Some(download.priority),
         };
         if self
+            .dynconfig
             .block_list
             .is_task_download_blocked(&check_params)
             .await
@@ -1157,6 +1156,7 @@ impl DfdaemonDownload for DfdaemonDownloadServerHandler {
             priority: None,
         };
         if self
+            .dynconfig
             .block_list
             .is_persistent_task_download_blocked(check_params)
             .await
@@ -1481,6 +1481,7 @@ impl DfdaemonDownload for DfdaemonDownloadServerHandler {
             tag: None,
         };
         if self
+            .dynconfig
             .block_list
             .is_persistent_task_upload_blocked(check_params)
             .await
@@ -1761,6 +1762,7 @@ impl DfdaemonDownload for DfdaemonDownloadServerHandler {
             priority: None,
         };
         if self
+            .dynconfig
             .block_list
             .is_persistent_cache_task_download_blocked(check_params)
             .await
@@ -2039,6 +2041,7 @@ impl DfdaemonDownload for DfdaemonDownloadServerHandler {
             tag: request.tag.clone(),
         };
         if self
+            .dynconfig
             .block_list
             .is_persistent_cache_task_upload_blocked(check_params)
             .await

dragonfly-client/src/grpc/mod.rs

@@ -24,7 +24,6 @@ use std::time::Duration;
 use tonic::Request;
 use tracing::{debug, error, info, instrument, Instrument};
 
-pub mod block_list;
 pub mod dfdaemon_download;
 pub mod dfdaemon_upload;
 pub mod health;