并行执行多个DADK任务 (#15)

Author: Jomocool

src/console/mod.rs

@@ -39,6 +39,10 @@ pub struct CommandLineArgs {
     /// DADK缓存根目录
     #[arg(long, value_parser = parse_check_dir_exists)]
     pub cache_dir: Option<PathBuf>,
+
+    /// DADK任务并行线程数量
+    #[arg(short, long)]
+    pub thread: Option<usize>,
 }
 
 /// @brief 检查目录是否存在

src/executor/cache.rs

@@ -1,4 +1,4 @@
-use std::{path::PathBuf, rc::Rc};
+use std::{path::PathBuf, sync::Arc};
 
 use log::info;
 
@@ -78,7 +78,7 @@ pub fn cache_root_init(path: Option<PathBuf>) -> Result<(), ExecutorError> {
 #[derive(Debug, Clone)]
 pub struct CacheDir {
     #[allow(dead_code)]
-    entity: Rc<SchedEntity>,
+    entity: Arc<SchedEntity>,
     pub path: PathBuf,
     pub cache_type: CacheDirType,
 }
@@ -92,9 +92,9 @@ pub enum CacheDirType {
 impl CacheDir {
     pub const DADK_BUILD_CACHE_DIR_ENV_KEY_PREFIX: &'static str = "DADK_BUILD_CACHE_DIR";
     pub const DADK_SOURCE_CACHE_DIR_ENV_KEY_PREFIX: &'static str = "DADK_SOURCE_CACHE_DIR";
-    pub fn new(entity: Rc<SchedEntity>, cache_type: CacheDirType) -> Result<Self, ExecutorError> {
+    pub fn new(entity: Arc<SchedEntity>, cache_type: CacheDirType) -> Result<Self, ExecutorError> {
         let task = entity.task();
-        let path = Self::get_path(task, cache_type);
+        let path = Self::get_path(&task, cache_type);
 
         let result = Self {
             entity,
@@ -122,15 +122,15 @@ impl CacheDir {
         return PathBuf::from(cache_dir);
     }
 
-    pub fn build_dir(entity: Rc<SchedEntity>) -> Result<PathBuf, ExecutorError> {
-        return Ok(Self::new(entity, CacheDirType::Build)?.path);
+    pub fn build_dir(entity: Arc<SchedEntity>) -> Result<PathBuf, ExecutorError> {
+        return Ok(Self::new(entity.clone(), CacheDirType::Build)?.path);
     }
 
-    pub fn source_dir(entity: Rc<SchedEntity>) -> Result<PathBuf, ExecutorError> {
-        return Ok(Self::new(entity, CacheDirType::Source)?.path);
+    pub fn source_dir(entity: Arc<SchedEntity>) -> Result<PathBuf, ExecutorError> {
+        return Ok(Self::new(entity.clone(), CacheDirType::Source)?.path);
     }
 
-    pub fn build_dir_env_key(entity: &Rc<SchedEntity>) -> Result<String, ExecutorError> {
+    pub fn build_dir_env_key(entity: &Arc<SchedEntity>) -> Result<String, ExecutorError> {
         let name_version_env = entity.task().name_version_env();
         return Ok(format!(
             "{}_{}",
@@ -139,7 +139,7 @@ impl CacheDir {
         ));
     }
 
-    pub fn source_dir_env_key(entity: &Rc<SchedEntity>) -> Result<String, ExecutorError> {
+    pub fn source_dir_env_key(entity: &Arc<SchedEntity>) -> Result<String, ExecutorError> {
         let name_version_env = entity.task().name_version_env();
         return Ok(format!(
             "{}_{}",
@@ -148,7 +148,7 @@ impl CacheDir {
         ));
     }
 
-    pub fn need_source_cache(entity: &Rc<SchedEntity>) -> bool {
+    pub fn need_source_cache(entity: &Arc<SchedEntity>) -> bool {
         let task_type = &entity.task().task_type;
 
         if let TaskType::BuildFromSource(cs) = task_type {

src/executor/mod.rs

@@ -3,8 +3,7 @@ use std::{
     env::Vars,
     path::PathBuf,
     process::{Command, Stdio},
-    rc::Rc,
-    sync::RwLock,
+    sync::{Arc, RwLock},
 };
 
 use log::{debug, error, info, warn};
@@ -30,7 +29,7 @@ lazy_static! {
 
 #[derive(Debug, Clone)]
 pub struct Executor {
-    entity: Rc<SchedEntity>,
+    entity: Arc<SchedEntity>,
     action: Action,
     local_envs: EnvMap,
     /// 任务构建结果输出到的目录
@@ -55,7 +54,7 @@ impl Executor {
     /// * `Ok(Executor)` - 创建成功
     /// * `Err(ExecutorError)` - 创建失败
     pub fn new(
-        entity: Rc<SchedEntity>,
+        entity: Arc<SchedEntity>,
         action: Action,
         dragonos_sysroot: PathBuf,
     ) -> Result<Self, ExecutorError> {
@@ -148,7 +147,8 @@ impl Executor {
 
     /// # 执行安装操作，把构建结果安装到DragonOS
     fn install(&self) -> Result<(), ExecutorError> {
-        let in_dragonos_path = self.entity.task().install.in_dragonos_path.as_ref();
+        let binding = self.entity.task();
+        let in_dragonos_path = binding.install.in_dragonos_path.as_ref();
         // 如果没有指定安装路径，则不执行安装
         if in_dragonos_path.is_none() {
             return Ok(());
@@ -336,7 +336,8 @@ impl Executor {
         // 设置本地环境变量
         self.prepare_target_env()?;
 
-        let task_envs: Option<&Vec<TaskEnv>> = self.entity.task().envs.as_ref();
+        let binding = self.entity.task();
+        let task_envs: Option<&Vec<TaskEnv>> = binding.envs.as_ref();
         if task_envs.is_none() {
             return Ok(());
         }
@@ -538,7 +539,7 @@ pub fn prepare_env(sched_entities: &SchedEntities) -> Result<(), ExecutorError>
     env_list.add_vars(envs);
 
     // 为每个任务创建特定的环境变量
-    for entity in sched_entities.iter() {
+    for entity in sched_entities.entities().iter() {
         // 导出任务的构建目录环境变量
         let build_dir = CacheDir::build_dir(entity.clone())?;
 

src/main.rs

@@ -83,6 +83,7 @@
 //! - 支持自动更新
 //! - 完善clean命令的逻辑
 
+#![feature(extract_if)]
 #![feature(io_error_more)]
 
 #[macro_use]
@@ -103,7 +104,7 @@ use simple_logger::SimpleLogger;
 use crate::{
     console::{interactive::InteractiveConsole, CommandLineArgs},
     executor::cache::cache_root_init,
-    scheduler::Scheduler,
+    scheduler::{task_deque::TASK_DEQUE, Scheduler},
 };
 
 mod console;
@@ -124,6 +125,7 @@ fn main() {
     let dragonos_dir = args.dragonos_dir.clone();
     let config_dir = args.config_dir.clone();
     let action = args.action;
+    let thread = args.thread;
     info!(
         "DragonOS sysroot dir: {}",
         dragonos_dir
@@ -137,6 +139,12 @@ fn main() {
             .map_or_else(|| "None".to_string(), |d| d.display().to_string())
     );
     info!("Action: {:?}", action);
+    info!(
+        "Thread num: {}",
+        thread
+            .as_ref()
+            .map_or_else(|| "None".to_string(), |d| d.to_string())
+    );
 
     match action {
         console::Action::New => {
@@ -150,6 +158,10 @@ fn main() {
         _ => {}
     }
 
+    if let Some(thread) = thread {
+        TASK_DEQUE.lock().unwrap().set_thread(thread);
+    }
+
     // 初始化缓存目录
     let r = cache_root_init(args.cache_dir);
     if r.is_err() {