From c9f73b1696752c4c9f9e68c8345d163f939d275d Mon Sep 17 00:00:00 2001 From: Ruihang Xia Date: Thu, 14 May 2026 17:44:26 +0800 Subject: [PATCH] feat: start environments in parallel (#8101) * start environments in parallel Signed-off-by: Ruihang Xia * use merged commit Signed-off-by: Ruihang Xia * fix bins_dir racing Signed-off-by: Ruihang Xia --------- Signed-off-by: Ruihang Xia --- Cargo.lock | 2 +- tests/runner/Cargo.toml | 2 +- tests/runner/src/cmd/bare.rs | 4 +- tests/runner/src/env/bare.rs | 73 +++++++++++++++++++++++++++--------- 4 files changed, 60 insertions(+), 21 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a9332267f2..d645781c98 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -13130,7 +13130,7 @@ dependencies = [ [[package]] name = "sqlness" version = "0.6.1" -source = "git+https://github.com/GreptimeTeam/sqlness?rev=6bc85ce077a87756e68ef5a5cd5707071a9da380#6bc85ce077a87756e68ef5a5cd5707071a9da380" +source = "git+https://github.com/GreptimeTeam/sqlness?rev=cf4c2ae005fb8531a023ea638124ef49f1022025#cf4c2ae005fb8531a023ea638124ef49f1022025" dependencies = [ "async-trait", "derive_builder 0.11.2", diff --git a/tests/runner/Cargo.toml b/tests/runner/Cargo.toml index 6b41e5a61e..75b62f6ff7 100644 --- a/tests/runner/Cargo.toml +++ b/tests/runner/Cargo.toml @@ -25,7 +25,7 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls" serde.workspace = true serde_json.workspace = true sha2 = "0.10" -sqlness = { git = "https://github.com/GreptimeTeam/sqlness", rev = "6bc85ce077a87756e68ef5a5cd5707071a9da380" } +sqlness = { git = "https://github.com/GreptimeTeam/sqlness", rev = "cf4c2ae005fb8531a023ea638124ef49f1022025" } tar = "0.4" tempfile.workspace = true tinytemplate = "1.2" diff --git a/tests/runner/src/cmd/bare.rs b/tests/runner/src/cmd/bare.rs index 58199f959e..f845d7b2c0 100644 --- a/tests/runner/src/cmd/bare.rs +++ b/tests/runner/src/cmd/bare.rs @@ -137,12 +137,13 @@ impl BareCommand { || self.setup_etcd || self.setup_pg.is_some() || self.setup_mysql.is_some() + || matches!(self.wal, Wal::Kafka) || self.kafka_wal_broker_endpoints.is_some() || self.config.test_filter != ".*" { self.jobs = 1; println!( - "Normalizing parallelism to 1 due to server addresses, etcd/pg/mysql setup, or test filter usage" + "Normalizing parallelism to 1 due to server addresses, etcd/pg/mysql/kafka setup, or test filter usage" ); } @@ -154,6 +155,7 @@ impl BareCommand { .env_config_file(self.config.env_config_file) .interceptor_registry(interceptor_registry) .parallelism(self.jobs) + .env_parallelism(self.jobs) .build() .unwrap(); diff --git a/tests/runner/src/env/bare.rs b/tests/runner/src/env/bare.rs index 1501d18512..61e93d26db 100644 --- a/tests/runner/src/env/bare.rs +++ b/tests/runner/src/env/bare.rs @@ -249,6 +249,7 @@ impl Env { metasrv_process: None.into(), frontend_process: None.into(), flownode_process: None.into(), + active_bins_dir: Mutex::new(self.bins_dir.lock().unwrap().clone()), ctx: GreptimeDBContext { time: 0, datanode_id: Default::default(), @@ -273,6 +274,22 @@ impl Env { db_ctx: &GreptimeDBContext, id: usize, truncate_log: bool, + ) -> Child { + let bins_dir = self.bins_dir.lock().unwrap().clone().expect( + "GreptimeDB binary is not available. Please pass in the path to the directory that contains the pre-built GreptimeDB binary. Or you may call `self.build_db()` beforehand.", + ); + + self.start_server_with_bins_dir(mode, db_ctx, id, truncate_log, bins_dir) + .await + } + + async fn start_server_with_bins_dir( + &self, + mode: ServerMode, + db_ctx: &GreptimeDBContext, + id: usize, + truncate_log: bool, + bins_dir: PathBuf, ) -> Child { let log_file_name = match mode { ServerMode::Datanode { node_id, .. } => { @@ -310,10 +327,6 @@ impl Env { let program = PROGRAM; - let bins_dir = self.bins_dir.lock().unwrap().clone().expect( - "GreptimeDB binary is not available. Please pass in the path to the directory that contains the pre-built GreptimeDB binary. Or you may call `self.build_db()` beforehand.", - ); - let abs_bins_dir = bins_dir .canonicalize() .expect("Failed to canonicalize bins_dir"); @@ -348,6 +361,10 @@ impl Env { /// stop and restart the server process async fn restart_server(&self, db: &GreptimeDB, is_full_restart: bool) { + let bins_dir = db.active_bins_dir.lock().unwrap().clone().expect( + "GreptimeDB binary is not available. Please pass in the path to the directory that contains the pre-built GreptimeDB binary. Or you may call `self.build_db()` beforehand.", + ); + { if let Some(server_process) = db.server_processes.clone() { let mut server_processes = server_process.lock().unwrap(); @@ -384,7 +401,9 @@ impl Env { .cloned() .unwrap(); let server_addr = server_mode.server_addr().unwrap(); - let new_server_process = self.start_server(server_mode, &db.ctx, db.id, false).await; + let new_server_process = self + .start_server_with_bins_dir(server_mode, &db.ctx, db.id, false, bins_dir.clone()) + .await; let mut client = db.client.lock().await; client @@ -402,7 +421,15 @@ impl Env { .get_server_mode(SERVER_MODE_METASRV_IDX) .cloned() .unwrap(); - let metasrv = self.start_server(metasrv_mode, &db.ctx, db.id, false).await; + let metasrv = self + .start_server_with_bins_dir( + metasrv_mode, + &db.ctx, + db.id, + false, + bins_dir.clone(), + ) + .await; db.metasrv_process .lock() .expect("lock poisoned") @@ -421,7 +448,13 @@ impl Env { .cloned() .unwrap(); let new_server_process = self - .start_server(datanode_mode, &db.ctx, db.id, false) + .start_server_with_bins_dir( + datanode_mode, + &db.ctx, + db.id, + false, + bins_dir.clone(), + ) .await; processes.push(new_server_process); } @@ -433,7 +466,13 @@ impl Env { .cloned() .unwrap(); let frontend = self - .start_server(frontend_mode, &db.ctx, db.id, false) + .start_server_with_bins_dir( + frontend_mode, + &db.ctx, + db.id, + false, + bins_dir.clone(), + ) .await; db.frontend_process .lock() @@ -447,7 +486,7 @@ impl Env { .cloned() .unwrap(); let flownode = self - .start_server(flownode_mode, &db.ctx, db.id, false) + .start_server_with_bins_dir(flownode_mode, &db.ctx, db.id, false, bins_dir.clone()) .await; db.flownode_process .lock() @@ -516,7 +555,8 @@ impl Env { /// Build the DB with `cargo build --bin greptime` fn build_db(&self) { - if self.bins_dir.lock().unwrap().is_some() { + let mut bins_dir = self.bins_dir.lock().unwrap(); + if bins_dir.is_some() { return; } @@ -541,11 +581,7 @@ impl Env { panic!(); } - let _ = self - .bins_dir - .lock() - .unwrap() - .insert(util::get_binary_dir("debug")); + bins_dir.replace(util::get_binary_dir("debug")); } pub(crate) fn extra_args(&self) -> &Vec { @@ -559,6 +595,7 @@ pub struct GreptimeDB { frontend_process: Mutex>, flownode_process: Mutex>, client: TokioMutex, + active_bins_dir: Mutex>, ctx: GreptimeDBContext, is_standalone: bool, env: Env, @@ -609,16 +646,16 @@ impl Database for GreptimeDB { .cloned(); match version_bin_dir { - Some(path) if path.clone().join(PROGRAM).is_file() => { + Some(path) if path.join(PROGRAM).is_file() => { // use version in versioned_bins_dirs - *self.env.bins_dir.lock().unwrap() = Some(path.clone()); + *self.active_bins_dir.lock().unwrap() = Some(path); } _ => { // use version in dir files maybe_pull_binary(version, self.env.pull_version_on_need).await; let root = get_workspace_root(); let new_path = PathBuf::from_iter([&root, version]); - *self.env.bins_dir.lock().unwrap() = Some(new_path); + *self.active_bins_dir.lock().unwrap() = Some(new_path); } }