starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-19 16:02:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: reduce stack size needed for workflow transitions

Browse Source
This commit is contained in:
Ruben Fiszel
2024-01-08 00:47:59 +01:00
parent bb84ab654a
commit f4ea893ad0
7 changed files with 417 additions and 408 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+44 -214
View File
@@ -539,15 +539,6 @@ dependencies = [
"pin-project-lite",
]
[[package]]
name = "atoi"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d7c57d12312ff59c811c0643f4d80830505833c9ffaebd193d819392b265be8e"
dependencies = [
"num-traits",
]
[[package]]
name = "atoi"
version = "2.0.0"
@@ -1073,7 +1064,7 @@ dependencies = [
"async-trait",
"futures-channel",
"futures-util",
"parking_lot 0.12.1",
"parking_lot",
"tokio",
]
@@ -1879,7 +1870,7 @@ dependencies = [
"bitflags 2.4.1",
"crossterm_winapi",
"libc",
"parking_lot 0.12.1",
"parking_lot",
"winapi",
]
@@ -2010,7 +2001,7 @@ dependencies = [
"hashbrown 0.14.3",
"lock_api",
"once_cell",
"parking_lot_core 0.9.9",
"parking_lot_core",
]
[[package]]
@@ -2129,7 +2120,7 @@ dependencies = [
"libc",
"log",
"once_cell",
"parking_lot 0.12.1",
"parking_lot",
"pin-project",
"serde",
"serde_json",
@@ -2352,22 +2343,13 @@ dependencies = [
"subtle",
]
[[package]]
name = "dirs"
version = "4.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ca3aa72a6f96ea37bbc5aa912f6788242832f75369bdfdadcb0e38423f100059"
dependencies = [
"dirs-sys 0.3.7",
]
[[package]]
name = "dirs"
version = "5.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "44c45a9d03d6676652bcb5e724c7e988de1acad23a711b5217ab9cbecbec2225"
dependencies = [
"dirs-sys 0.4.1",
"dirs-sys",
]
[[package]]
@@ -2380,17 +2362,6 @@ dependencies = [
"dirs-sys-next",
]
[[package]]
name = "dirs-sys"
version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b1d1d91c932ef41c0f2663aa8b0ca0342d444d842c06914aa0a7e352d0bada6"
dependencies = [
"libc",
"redox_users",
"winapi",
]
[[package]]
name = "dirs-sys"
version = "0.4.1"
@@ -2945,17 +2916,6 @@ dependencies = [
"futures-util",
]
[[package]]
name = "futures-intrusive"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a604f7a68fbf8103337523b1fadc8ade7361ee3f112f7c680ad179651616aed5"
dependencies = [
"futures-core",
"lock_api",
"parking_lot 0.11.2",
]
[[package]]
name = "futures-intrusive"
version = "0.5.0"
@@ -2964,7 +2924,7 @@ checksum = "1d930c203dd0b6ff06e0201a4a2fe9149b43c684fd4420555b26d21b1a02956f"
dependencies = [
"futures-core",
"lock_api",
"parking_lot 0.12.1",
"parking_lot",
]
[[package]]
@@ -3439,7 +3399,7 @@ version = "0.3.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2b780635574b3d92f036890d8373433d6f9fc7abb320ee42a5c25897fc8ed732"
dependencies = [
"dirs 5.0.1",
"dirs",
"indicatif",
"log",
"native-tls",
@@ -3910,9 +3870,9 @@ checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55"
[[package]]
name = "libc"
version = "0.2.151"
version = "0.2.152"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "302d7ab3130588088d277783b1e2d2e10c9e9e4a16dd9050e6ec93fb3e7048f4"
checksum = "13e3bf6590cbc649f4d1a3eefc9d5d6eb746f5200ffb04e5e142700b8faa56e7"
[[package]]
name = "libgit2-sys"
@@ -4614,7 +4574,7 @@ dependencies = [
"humantime",
"hyper",
"itertools 0.11.0",
"parking_lot 0.12.1",
"parking_lot",
"percent-encoding",
"quick-xml 0.31.0",
"rand 0.8.5",
@@ -4753,17 +4713,6 @@ version = "2.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bb813b8af86854136c6922af0598d719255ecb2179515e6e7730d468f05c9cae"
[[package]]
name = "parking_lot"
version = "0.11.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99"
dependencies = [
"instant",
"lock_api",
"parking_lot_core 0.8.6",
]
[[package]]
name = "parking_lot"
version = "0.12.1"
@@ -4771,21 +4720,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3742b2c103b9f06bc9fff0a37ff4912935851bee6d36f3c02bcc755bcfec228f"
dependencies = [
"lock_api",
"parking_lot_core 0.9.9",
]
[[package]]
name = "parking_lot_core"
version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "60a2cfe6f0ad2bfc16aefa463b497d5c7a5ecd44a23efa72aa342d90177356dc"
dependencies = [
"cfg-if",
"instant",
"libc",
"redox_syscall 0.2.16",
"smallvec",
"winapi",
"parking_lot_core",
]
[[package]]
@@ -4923,14 +4858,11 @@ dependencies = [
"archiver-rs",
"async-trait",
"bytes",
"dirs 5.0.1",
"dirs",
"futures",
"lazy_static",
"log",
"reqwest",
"sqlx 0.6.3",
"thiserror",
"tokio",
"zip",
]
@@ -5177,7 +5109,7 @@ checksum = "dd503430a6d9779b07915d858865fe998317ef3cfef8973881f578ac5d4baae7"
dependencies = [
"ahash 0.8.7",
"arrow-format",
"atoi 2.0.0",
"atoi",
"atoi_simd",
"bytemuck",
"chrono",
@@ -5445,7 +5377,7 @@ version = "0.35.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "aae56f79e9cedd617773c1c8f5ca84a31a8b1d593714959d5f799e7bdd98fe51"
dependencies = [
"atoi 2.0.0",
"atoi",
"chrono",
"now",
"once_cell",
@@ -5764,7 +5696,7 @@ dependencies = [
"fnv",
"lazy_static",
"memchr",
"parking_lot 0.12.1",
"parking_lot",
"thiserror",
]
@@ -5870,7 +5802,7 @@ dependencies = [
"ahash 0.8.7",
"equivalent",
"hashbrown 0.14.3",
"parking_lot 0.12.1",
"parking_lot",
]
[[package]]
@@ -6043,15 +5975,6 @@ dependencies = [
"url",
]
[[package]]
name = "redox_syscall"
version = "0.2.16"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fb5a58c1855b4b6819d59012155603f0b22ad30cad752600aadfcb695265519a"
dependencies = [
"bitflags 1.3.2",
]
[[package]]
name = "redox_syscall"
version = "0.3.5"
@@ -7294,82 +7217,19 @@ dependencies = [
"log",
]
[[package]]
name = "sqlx"
version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f8de3b03a925878ed54a954f621e64bf55a3c1bd29652d0d1a17830405350188"
dependencies = [
"sqlx-core 0.6.3",
"sqlx-macros 0.6.3",
]
[[package]]
name = "sqlx"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dba03c279da73694ef99763320dea58b51095dfe87d001b1d4b5fe78ba8763cf"
dependencies = [
"sqlx-core 0.7.3",
"sqlx-macros 0.7.3",
"sqlx-core",
"sqlx-macros",
"sqlx-mysql",
"sqlx-postgres",
"sqlx-sqlite",
]
[[package]]
name = "sqlx-core"
version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fa8241483a83a3f33aa5fff7e7d9def398ff9990b2752b6c6112b83c6d246029"
dependencies = [
"ahash 0.7.7",
"atoi 1.0.0",
"base64 0.13.1",
"bitflags 1.3.2",
"byteorder",
"bytes",
"crc",
"crossbeam-queue",
"dirs 4.0.0",
"dotenvy",
"either",
"event-listener 2.5.3",
"futures-channel",
"futures-core",
"futures-intrusive 0.4.2",
"futures-util",
"hashlink",
"hex",
"hkdf",
"hmac",
"indexmap 1.9.3",
"itoa",
"libc",
"log",
"md-5 0.10.6",
"memchr",
"once_cell",
"paste",
"percent-encoding",
"rand 0.8.5",
"rustls 0.20.9",
"rustls-pemfile",
"serde",
"serde_json",
"sha1",
"sha2 0.10.8",
"smallvec",
"sqlformat",
"sqlx-rt",
"stringprep",
"thiserror",
"tokio-stream",
"url",
"webpki-roots 0.22.6",
"whoami",
]
[[package]]
name = "sqlx-core"
version = "0.7.3"
@@ -7377,7 +7237,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d84b0a3c3739e220d94b3239fd69fb1f74bc36e16643423bd99de3b43c21bfbd"
dependencies = [
"ahash 0.8.7",
"atoi 2.0.0",
"atoi",
"bigdecimal 0.3.1",
"byteorder",
"bytes",
@@ -7389,7 +7249,7 @@ dependencies = [
"event-listener 2.5.3",
"futures-channel",
"futures-core",
"futures-intrusive 0.5.0",
"futures-intrusive",
"futures-io",
"futures-util",
"hashlink",
@@ -7416,25 +7276,6 @@ dependencies = [
"webpki-roots 0.25.3",
]
[[package]]
name = "sqlx-macros"
version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "9966e64ae989e7e575b19d7265cb79d7fc3cbbdf179835cb0d716f294c2049c9"
dependencies = [
"dotenvy",
"either",
"heck",
"once_cell",
"proc-macro2",
"quote",
"sha2 0.10.8",
"sqlx-core 0.6.3",
"sqlx-rt",
"syn 1.0.109",
"url",
]
[[package]]
name = "sqlx-macros"
version = "0.7.3"
@@ -7443,7 +7284,7 @@ checksum = "89961c00dc4d7dffb7aee214964b065072bff69e36ddb9e2c107541f75e4f2a5"
dependencies = [
"proc-macro2",
"quote",
"sqlx-core 0.7.3",
"sqlx-core",
"sqlx-macros-core",
"syn 1.0.109",
]
@@ -7465,7 +7306,7 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.10.8",
"sqlx-core 0.7.3",
"sqlx-core",
"sqlx-mysql",
"sqlx-postgres",
"sqlx-sqlite",
@@ -7481,7 +7322,7 @@ version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e37195395df71fd068f6e2082247891bc11e3289624bbc776a0cdfa1ca7f1ea4"
dependencies = [
"atoi 2.0.0",
"atoi",
"base64 0.21.5",
"bigdecimal 0.3.1",
"bitflags 2.4.1",
@@ -7512,7 +7353,7 @@ dependencies = [
"sha1",
"sha2 0.10.8",
"smallvec",
"sqlx-core 0.7.3",
"sqlx-core",
"stringprep",
"thiserror",
"tracing",
@@ -7526,7 +7367,7 @@ version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d6ac0ac3b7ccd10cc96c7ab29791a7dd236bd94021f31eec7ba3d46a74aa1c24"
dependencies = [
"atoi 2.0.0",
"atoi",
"base64 0.21.5",
"bigdecimal 0.3.1",
"bitflags 2.4.1",
@@ -7555,7 +7396,7 @@ dependencies = [
"sha1",
"sha2 0.10.8",
"smallvec",
"sqlx-core 0.7.3",
"sqlx-core",
"stringprep",
"thiserror",
"tracing",
@@ -7563,36 +7404,25 @@ dependencies = [
"whoami",
]
[[package]]
name = "sqlx-rt"
version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "804d3f245f894e61b1e6263c84b23ca675d96753b5abfd5cc8597d86806e8024"
dependencies = [
"once_cell",
"tokio",
"tokio-rustls 0.23.4",
]
[[package]]
name = "sqlx-sqlite"
version = "0.7.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "210976b7d948c7ba9fced8ca835b11cbb2d677c59c79de41ac0d397e14547490"
dependencies = [
"atoi 2.0.0",
"atoi",
"chrono",
"flume",
"futures-channel",
"futures-core",
"futures-executor",
"futures-intrusive 0.5.0",
"futures-intrusive",
"futures-util",
"libsqlite3-sys",
"log",
"percent-encoding",
"serde",
"sqlx-core 0.7.3",
"sqlx-core",
"tracing",
"url",
"urlencoding",
@@ -7653,7 +7483,7 @@ checksum = "f91138e76242f575eb1d3b38b4f1362f10d3a43f47d182a5b359af488a02293b"
dependencies = [
"new_debug_unreachable",
"once_cell",
"parking_lot 0.12.1",
"parking_lot",
"phf_shared 0.10.0",
"precomputed-hash",
"serde",
@@ -7697,9 +7527,9 @@ dependencies = [
[[package]]
name = "strsim"
version = "0.10.0"
version = "0.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "73473c0e59e6d5812c5dfe2a064a6444949f089e20eec9a2e5506596494e4623"
checksum = "ccbca6f34534eb78dbee83f6b2c9442fea7113f43d9e80ea320f0972ae5dc08d"
[[package]]
name = "strum"
@@ -8399,7 +8229,7 @@ dependencies = [
"libc",
"mio",
"num_cpus",
"parking_lot 0.12.1",
"parking_lot",
"pin-project-lite",
"signal-hook-registry",
"socket2 0.5.5",
@@ -8453,7 +8283,7 @@ dependencies = [
"futures-channel",
"futures-util",
"log",
"parking_lot 0.12.1",
"parking_lot",
"percent-encoding",
"phf 0.11.2",
"pin-project-lite",
@@ -8661,7 +8491,7 @@ dependencies = [
"cookie",
"futures-util",
"http",
"parking_lot 0.12.1",
"parking_lot",
"pin-project-lite",
"tower-layer",
"tower-service",
@@ -9462,7 +9292,7 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.10.8",
"sqlx 0.7.3",
"sqlx",
"tokio",
"tokio-metrics",
"tracing",
@@ -9525,7 +9355,7 @@ dependencies = [
"serde_urlencoded",
"sha2 0.10.8",
"sql-builder",
"sqlx 0.7.3",
"sqlx",
"tempfile",
"time",
"tinyvector",
@@ -9574,7 +9404,7 @@ dependencies = [
"serde",
"serde_json",
"sql-builder",
"sqlx 0.7.3",
"sqlx",
"tracing",
"windmill-common",
]
@@ -9602,7 +9432,7 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.10.8",
"sqlx 0.7.3",
"sqlx",
"thiserror",
"tokio",
"tracing",
@@ -9614,12 +9444,12 @@ dependencies = [
[[package]]
name = "windmill-git-sync"
version = "1.235.0"
version = "1.235.1"
dependencies = [
"rsmq_async",
"serde",
"serde_json",
"sqlx 0.7.3",
"sqlx",
"tracing",
"windmill-common",
"windmill-queue",
@@ -9690,7 +9520,7 @@ dependencies = [
"regex",
"rustpython-parser",
"serde_json",
"sqlx 0.7.3",
"sqlx",
"windmill-common",
"windmill-parser",
]
@@ -9764,7 +9594,7 @@ dependencies = [
"serde_json",
"serde_urlencoded",
"sql-builder",
"sqlx 0.7.3",
"sqlx",
"tokio",
"tracing",
"ulid",
@@ -9815,7 +9645,7 @@ dependencies = [
"serde",
"serde_json",
"sha2 0.10.8",
"sqlx 0.7.3",
"sqlx",
"tiberius",
"tokio",
"tokio-postgres",
backend/Cargo.toml
+1 -1
View File
@@ -68,7 +68,7 @@ uuid.workspace = true
gethostname.workspace = true
serde_json.workspace = true
serde.workspace = true
pg-embed = {git = "https://github.com/faokunega/pg-embed", optional = true}
pg-embed = {git = "https://github.com/faokunega/pg-embed", optional = true, default-features = false}
[dev-dependencies]
serde_json.workspace = true
backend/src/monitor.rs
+4 -2
View File
@@ -35,8 +35,8 @@ use windmill_common::{
BASE_URL, DB, METRICS_DEBUG_ENABLED, METRICS_ENABLED,
};
use windmill_worker::{
create_token_for_owner, handle_job_error, AuthedClient, JOB_DEFAULT_TIMEOUT, KEEP_JOB_DIR,
NPM_CONFIG_REGISTRY, PIP_EXTRA_INDEX_URL, SCRIPT_TOKEN_EXPIRY,
create_token_for_owner, handle_job_error, AuthedClient, SendResult, JOB_DEFAULT_TIMEOUT,
KEEP_JOB_DIR, NPM_CONFIG_REGISTRY, PIP_EXTRA_INDEX_URL, SCRIPT_TOKEN_EXPIRY,
};
#[cfg(feature = "enterprise")]
@@ -757,6 +757,7 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
// since the job is unrecoverable, the same worker queue should never be sent anything
let (same_worker_tx_never_used, _same_worker_rx_never_used) = mpsc::channel::<Uuid>(1);
let (send_result_never_used, _send_result_rx_never_used) = mpsc::channel::<SendResult>(1);
let token = create_token_for_owner(
&db,
@@ -795,6 +796,7 @@ async fn handle_zombie_jobs<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
"",
rsmq.clone(),
worker_name,
send_result_never_used,
)
.await;
}
backend/tests/worker.rs
+1 -1
View File
@@ -1805,7 +1805,7 @@ def main():
}
#[sqlx::test(fixtures("base"))]
async fn test_empty_loop(db: Pool<Postgres>) {
async fn test_empty_loop_1(db: Pool<Postgres>) {
initialize_tracing().await;
let server = ApiServer::start(db.clone()).await;
let port = server.addr.port();
backend/windmill-worker/src/worker.rs
+194 -108
View File
@@ -475,6 +475,7 @@ async fn handle_receive_completed_job<
worker_name: &str,
worker_save_completed_job_duration: Option<Arc<prometheus::Histogram>>,
worker_flow_transition_duration: Option<Arc<prometheus::Histogram>>,
job_completed_tx: Sender<SendResult>,
) {
let token = jc.token.clone();
let workspace = jc.job.workspace_id.clone();
@@ -497,6 +498,7 @@ async fn handle_receive_completed_job<
worker_name,
worker_save_completed_job_duration,
worker_flow_transition_duration,
job_completed_tx.clone(),
)
.await
{
@@ -512,6 +514,7 @@ async fn handle_receive_completed_job<
&worker_dir,
rsmq.clone(),
worker_name,
job_completed_tx,
)
.await;
}
@@ -519,7 +522,7 @@ async fn handle_receive_completed_job<
#[derive(Clone)]
pub struct JobCompletedSender(
Sender<JobCompleted>,
Sender<SendResult>,
Option<Arc<GenericGauge<AtomicI64>>>,
Option<Arc<prometheus::Histogram>>,
);
@@ -528,12 +531,12 @@ impl JobCompletedSender {
pub async fn send(
&self,
jc: JobCompleted,
) -> Result<(), tokio::sync::mpsc::error::SendError<JobCompleted>> {
) -> Result<(), tokio::sync::mpsc::error::SendError<SendResult>> {
if let Some(wj) = self.1.as_ref() {
wj.inc()
}
let timer = self.2.as_ref().map(|x| x.start_timer());
let r = self.0.send(jc).await;
let r = self.0.send(SendResult::JobCompleted(jc)).await;
timer.map(|x| x.stop_and_record());
r
}
@@ -879,7 +882,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let (same_worker_tx, mut same_worker_rx) = mpsc::channel::<Uuid>(5);
let (job_completed_tx, mut job_completed_rx) = mpsc::channel::<JobCompleted>(3);
let (job_completed_tx, mut job_completed_rx) = mpsc::channel::<SendResult>(3);
let job_completed_tx = JobCompletedSender(
job_completed_tx,
@@ -967,119 +970,167 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let worker_name2 = worker_name.clone();
let killpill_tx2 = killpill_tx.clone();
let job_completed_sender = job_completed_tx.0.clone();
let send_result = tokio::spawn(async move {
while let Some(jc) = job_completed_rx.recv().await {
if let Some(wj) = worker_job_completed_channel_queue2.as_ref() {
wj.dec();
}
let base_internal_url2 = base_internal_url2.clone();
let worker_dir2 = worker_dir2.clone();
let db2 = db2.clone();
let same_worker_tx2 = same_worker_tx2.clone();
let rsmq2 = rsmq2.clone();
let worker_name = worker_name2.clone();
if matches!(jc.job.job_kind, JobKind::Noop) || is_dedicated_worker {
thread_count.fetch_add(1, Ordering::SeqCst);
let thread_count = thread_count.clone();
loop {
if thread_count.load(Ordering::Relaxed) < 4 {
break;
while let Some(sr) = job_completed_rx.recv().await {
match sr {
SendResult::JobCompleted(jc) => {
if let Some(wj) = worker_job_completed_channel_queue2.as_ref() {
wj.dec();
}
tokio::time::sleep(Duration::from_millis(3)).await;
}
let base_internal_url2 = base_internal_url2.clone();
let worker_dir2 = worker_dir2.clone();
let db2 = db2.clone();
let same_worker_tx2 = same_worker_tx2.clone();
let rsmq2 = rsmq2.clone();
let worker_name = worker_name2.clone();
if matches!(jc.job.job_kind, JobKind::Noop) || is_dedicated_worker {
thread_count.fetch_add(1, Ordering::SeqCst);
let thread_count = thread_count.clone();
#[cfg(feature = "benchmark")]
let send_duration = send_duration2.clone();
#[cfg(feature = "benchmark")]
let process_duration = process_duration.clone();
#[cfg(feature = "benchmark")]
let completed_jobs = completed_jobs.clone();
#[cfg(feature = "benchmark")]
let main_duration = main_duration2.clone();
let worker_save_completed_job_duration2 =
worker_save_completed_job_duration.clone();
let worker_flow_transition_duration2 = worker_flow_transition_duration.clone();
let killpill_tx = killpill_tx2.clone();
tokio::spawn(async move {
#[cfg(feature = "benchmark")]
let process_start = Instant::now();
let is_dependency_job = matches!(
jc.job.job_kind,
JobKind::Dependencies | JobKind::FlowDependencies
);
handle_receive_completed_job(
jc,
base_internal_url2,
db2.clone(),
worker_dir2,
same_worker_tx2,
rsmq2,
&worker_name,
worker_save_completed_job_duration2.clone(),
worker_flow_transition_duration2.clone(),
)
.await;
#[cfg(feature = "benchmark")]
{
let n = completed_jobs.fetch_add(1, Ordering::SeqCst);
if (n + 1) % 1000 == 0 || n == (jobs - 1) as usize {
let duration_s = start.elapsed().as_secs_f64();
let jobs_per_sec = n as f64 / duration_s;
tracing::info!(
"completed {} jobs in {}s, {} jobs/s",
n + 1,
duration_s,
jobs_per_sec
);
tracing::info!(
"main loop without send {}s",
main_duration.load(Ordering::SeqCst) as f64 / 1000.0
);
tracing::info!(
"send job completed / send dedicated job duration {}s",
send_duration.load(Ordering::SeqCst) as f64 / 1000.0
);
tracing::info!(
"job completed process duration {}s",
process_duration.load(Ordering::SeqCst) as f64 / 1000.0
);
loop {
if thread_count.load(Ordering::Relaxed) < 4 {
break;
}
tokio::time::sleep(Duration::from_millis(3)).await;
}
process_duration.fetch_add(
process_start.elapsed().as_millis() as usize,
Ordering::SeqCst,
);
}
#[cfg(feature = "benchmark")]
let send_duration = send_duration2.clone();
#[cfg(feature = "benchmark")]
let process_duration = process_duration.clone();
#[cfg(feature = "benchmark")]
let completed_jobs = completed_jobs.clone();
#[cfg(feature = "benchmark")]
let main_duration = main_duration2.clone();
thread_count.fetch_sub(1, Ordering::SeqCst);
if is_dependency_job && is_dedicated_worker {
tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted.");
sqlx::query!("UPDATE config SET config = config WHERE name = $1", format!("worker__{}", *WORKER_GROUP))
let worker_save_completed_job_duration2 =
worker_save_completed_job_duration.clone();
let worker_flow_transition_duration2 =
worker_flow_transition_duration.clone();
let killpill_tx = killpill_tx2.clone();
let job_completed_sender = job_completed_sender.clone();
tokio::spawn(async move {
#[cfg(feature = "benchmark")]
let process_start = Instant::now();
let is_dependency_job = matches!(
jc.job.job_kind,
JobKind::Dependencies | JobKind::FlowDependencies
);
handle_receive_completed_job(
jc,
base_internal_url2,
db2.clone(),
worker_dir2,
same_worker_tx2,
rsmq2,
&worker_name,
worker_save_completed_job_duration2.clone(),
worker_flow_transition_duration2.clone(),
job_completed_sender.clone(),
)
.await;
#[cfg(feature = "benchmark")]
{
let n = completed_jobs.fetch_add(1, Ordering::SeqCst);
if (n + 1) % 1000 == 0 || n == (jobs - 1) as usize {
let duration_s = start.elapsed().as_secs_f64();
let jobs_per_sec = n as f64 / duration_s;
tracing::info!(
"completed {} jobs in {}s, {} jobs/s",
n + 1,
duration_s,
jobs_per_sec
);
tracing::info!(
"main loop without send {}s",
main_duration.load(Ordering::SeqCst) as f64 / 1000.0
);
tracing::info!(
"send job completed / send dedicated job duration {}s",
send_duration.load(Ordering::SeqCst) as f64 / 1000.0
);
tracing::info!(
"job completed process duration {}s",
process_duration.load(Ordering::SeqCst) as f64 / 1000.0
);
}
process_duration.fetch_add(
process_start.elapsed().as_millis() as usize,
Ordering::SeqCst,
);
}
thread_count.fetch_sub(1, Ordering::SeqCst);
if is_dependency_job && is_dedicated_worker {
tracing::error!("Dedicated worker executed a dependency job, a new script has been deployed. Exiting expecting to be restarted.");
sqlx::query!("UPDATE config SET config = config WHERE name = $1", format!("worker__{}", *WORKER_GROUP))
.execute(&db2)
.await
.expect("update config to trigger restart of all dedicated workers at that config");
killpill_tx.send(()).unwrap_or_default();
killpill_tx.send(()).unwrap_or_default();
}
});
} else {
handle_receive_completed_job(
jc,
base_internal_url2,
db2,
worker_dir2,
same_worker_tx2,
rsmq2,
&worker_name,
worker_save_completed_job_duration2.clone(),
worker_flow_transition_duration2.clone(),
job_completed_sender.clone(),
)
.await;
}
});
} else {
handle_receive_completed_job(
jc,
base_internal_url2,
db2,
worker_dir2,
same_worker_tx2,
rsmq2,
&worker_name,
worker_save_completed_job_duration2.clone(),
worker_flow_transition_duration2.clone(),
)
.await;
}
SendResult::UpdateFlow {
flow,
w_id,
success,
result,
worker_dir,
stop_early_override,
token,
} => {
// let r;
if let Err(e) = update_flow_status_after_job_completion(
&db2,
&AuthedClient {
base_internal_url: base_internal_url2.to_string(),
workspace: w_id.clone(),
token: token.clone(),
force_client: None,
},
flow,
&Uuid::nil(),
&w_id,
success,
&result,
true,
same_worker_tx2.clone(),
&worker_dir,
stop_early_override,
rsmq2.clone(),
&worker_name2,
job_completed_sender.clone(),
)
.await
{
tracing::error!("Error updating flow status after job completion: {e}");
}
}
SendResult::Kill => {
break;
}
}
}
@@ -1336,6 +1387,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
}
}
println!("received killpill for worker {}", i_worker);
job_completed_tx.0.send(SendResult::Kill).await.unwrap();
break
},
_ = copy_to_bucket_rx.recv() => {
@@ -1603,6 +1655,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
&worker_dir,
rsmq.clone(),
&worker_name,
(&job_completed_tx.0).clone(),
)
.await;
};
@@ -2086,6 +2139,7 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
worker_name: &str,
worker_save_completed_job_duration: Option<Arc<prometheus::Histogram>>,
worker_flow_transition_duration: Option<Arc<prometheus::Histogram>>,
job_completed_tx: Sender<SendResult>,
) -> windmill_common::error::Result<()> {
if success {
// println!("bef completed job{:?}", SystemTime::now());
@@ -2129,6 +2183,7 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
None,
rsmq.clone(),
worker_name,
job_completed_tx,
)
.await?;
timer.map(|x| x.stop_and_record());
@@ -2164,6 +2219,7 @@ pub async fn process_completed_job<R: rsmq_async::RsmqConnection + Send + Sync +
None,
rsmq,
worker_name,
job_completed_tx,
)
.await?;
}
@@ -2219,6 +2275,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
worker_dir: &str,
rsmq: Option<R>,
worker_name: &str,
job_completed_tx: Sender<SendResult>,
) {
let err = match err {
Error::JsonErr(err) => err,
@@ -2266,6 +2323,7 @@ pub async fn handle_job_error<R: rsmq_async::RsmqConnection + Send + Sync + Clon
None,
rsmq.clone(),
worker_name,
job_completed_tx.clone(),
)
.await;
@@ -2308,6 +2366,34 @@ fn extract_error_value(log_lines: &str, i: i32) -> Box<RawValue> {
);
}
pub enum SendResult {
JobCompleted(JobCompleted),
UpdateFlow {
flow: Uuid,
w_id: String,
success: bool,
result: Box<RawValue>,
worker_dir: String,
stop_early_override: Option<bool>,
token: String,
},
Kill,
}
// db: &DB,
// client: &AuthedClient,
// flow: uuid::Uuid,
// job_id_for_status: &Uuid,
// w_id: &str,
// success: bool,
// result: &'a RawValue,
// unrecoverable: bool,
// same_worker_tx: Sender<Uuid>,
// worker_dir: &str,
// stop_early_override: Option<bool>,
// rsmq: Option<R>,
// worker_name: &str,
#[derive(Debug, Clone)]
pub struct JobCompleted {
pub job: Arc<QueuedJob>,
@@ -2470,7 +2556,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
same_worker_tx,
worker_dir,
rsmq,
worker_name,
job_completed_tx.0.clone(),
)
.await?;
timer.map(|x| x.stop_and_record());
backend/windmill-worker/src/worker_flow.rs
+171 -81
View File
@@ -15,9 +15,8 @@ use std::time::Duration;
use crate::common::{hash_args, save_in_cache};
use crate::js_eval::{eval_timeout, IdContext};
use crate::{AuthedClient, PreviousResult, KEEP_JOB_DIR};
use crate::{AuthedClient, PreviousResult, SendResult, KEEP_JOB_DIR};
use anyhow::Context;
use async_recursion::async_recursion;
use serde::Serialize;
use serde_json::value::RawValue;
use serde_json::{json, Value};
@@ -70,8 +69,10 @@ pub async fn update_flow_status_after_job_completion<
stop_early_override: Option<bool>,
rsmq: Option<R>,
worker_name: &str,
job_completed_tx: Sender<SendResult>,
) -> error::Result<()> {
// this is manual tailrecursion because async_recursion blows up the stack
// todo!();
let mut rec = update_flow_status_after_job_completion_internal(
db,
client,
@@ -87,6 +88,7 @@ pub async fn update_flow_status_after_job_completion<
false,
rsmq.clone(),
worker_name,
job_completed_tx.clone(),
)
.await?;
while let Some(nrec) = rec {
@@ -105,6 +107,7 @@ pub async fn update_flow_status_after_job_completion<
nrec.skip_error_handler,
rsmq.clone(),
worker_name,
job_completed_tx.clone(),
)
.await
{
@@ -125,6 +128,7 @@ pub async fn update_flow_status_after_job_completion<
nrec.skip_error_handler,
rsmq.clone(),
worker_name,
job_completed_tx.clone(),
)
.await?
}
@@ -167,6 +171,7 @@ pub async fn update_flow_status_after_job_completion_internal<
skip_error_handler: bool,
rsmq: Option<R>,
worker_name: &str,
job_completed_tx: Sender<SendResult>,
) -> error::Result<Option<RecUpdateFlowStatusAfterJobCompletion>> {
let (
should_continue_flow,
@@ -575,6 +580,8 @@ pub async fn update_flow_status_after_job_completion_internal<
}
tx.commit().await?;
tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow status updated");
(
should_continue_flow,
flow_job,
@@ -686,6 +693,7 @@ pub async fn update_flow_status_after_job_completion_internal<
}
true
} else {
tracing::debug!(id = %flow_job.id, "start handle flow");
match handle_flow(
&flow_job,
db,
@@ -694,7 +702,7 @@ pub async fn update_flow_status_after_job_completion_internal<
same_worker_tx.clone(),
worker_dir,
rsmq.clone(),
worker_name,
job_completed_tx,
)
.await
{
@@ -1037,7 +1045,7 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
rsmq: Option<R>,
worker_name: &str,
job_completed_tx: Sender<SendResult>,
) -> anyhow::Result<()> {
let flow = flow_job
.parse_raw_flow()
@@ -1056,7 +1064,7 @@ pub async fn handle_flow<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
same_worker_tx,
worker_dir,
rsmq,
worker_name,
job_completed_tx,
)
.await?;
Ok(())
@@ -1089,7 +1097,7 @@ pub struct RawArgs {
pub args: Option<Json<HashMap<String, Box<RawValue>>>>,
}
#[async_recursion]
// #[async_recursion]
// #[instrument(level = "trace", skip_all)]
async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
flow_job: &QueuedJob,
@@ -1101,7 +1109,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
same_worker_tx: Sender<Uuid>,
worker_dir: &str,
rsmq: Option<R>,
worker_name: &str,
job_completed_tx: Sender<SendResult>,
) -> error::Result<()> {
let job_root = flow_job
.root_job
@@ -1122,29 +1130,51 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
// if this is an empty module of if the module has already been completed, successfully, update the parent flow
if flow.modules.is_empty() || matches!(status_module, FlowStatusModule::Success { .. }) {
let r;
return update_flow_status_after_job_completion(
db,
client,
flow_job.id,
&Uuid::nil(),
flow_job.workspace_id.as_str(),
true,
if flow.modules.is_empty() {
r = to_raw_value(&flow_job_args);
&r
} else {
// it has to be an empty for loop event
serde_json::from_str("[]").unwrap()
},
true,
same_worker_tx,
worker_dir,
None,
rsmq,
worker_name,
)
.await;
job_completed_tx
.send(SendResult::UpdateFlow {
flow: flow_job.id,
success: true,
result: if flow.modules.is_empty() {
to_raw_value(&flow_job_args)
} else {
// it has to be an empty for loop event
serde_json::from_str("[]").unwrap()
},
stop_early_override: None,
w_id: flow_job.workspace_id.clone(),
worker_dir: worker_dir.to_string(),
token: client.token.clone(),
})
.await
.map_err(|e| {
Error::InternalErr(format!(
"error sending update flow message to job completed channel: {e}"
))
})?;
// let r;
// return update_flow_status_after_job_completion(
// db,
// client,
// flow_job.id,
// &Uuid::nil(),
// flow_job.workspace_id.as_str(),
// true,
// if flow.modules.is_empty() {
// r = to_raw_value(&flow_job_args);
// &r
// } else {
// // it has to be an empty for loop event
// serde_json::from_str("[]").unwrap()
// },
// true,
// same_worker_tx,
// worker_dir,
// None,
// rsmq,
// worker_name,
// )
// .await;
return Ok(());
}
let arc_flow_job_args = Arc::new(flow_job_args.clone());
@@ -1166,25 +1196,45 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
flow_job.id
).fetch_one(db).await?.unwrap_or(0);
if count > 0 {
return update_flow_status_after_job_completion(
db,
client,
flow_job.id,
&Uuid::nil(),
flow_job.workspace_id.as_str(),
true,
serde_json::from_str(
"\"not allowed to overlap, scheduling next iteration\"",
)
.unwrap(),
true,
same_worker_tx,
worker_dir,
Some(true),
rsmq,
worker_name,
)
.await;
job_completed_tx
.send(SendResult::UpdateFlow {
flow: flow_job.id,
success: true,
result: serde_json::from_str(
"\"not allowed to overlap, scheduling next iteration\"",
)
.unwrap(),
stop_early_override: Some(true),
w_id: flow_job.workspace_id.clone(),
worker_dir: worker_dir.to_string(),
token: client.token.clone(),
})
.await
.map_err(|e| {
Error::InternalErr(format!(
"error sending update flow message to job completed channel: {e}"
))
})?;
// return update_flow_status_after_job_completion(
// db,
// client,
// flow_job.id,
// &Uuid::nil(),
// flow_job.workspace_id.as_str(),
// true,
// serde_json::from_str(
// "\"not allowed to overlap, scheduling next iteration\"",
// )
// .unwrap(),
// true,
// same_worker_tx,
// worker_dir,
// Some(true),
// rsmq,
// worker_name,
// )
// .await;
return Ok(());
}
}
}
@@ -1199,22 +1249,40 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
)
.await?;
if skip {
return update_flow_status_after_job_completion(
db,
client,
flow_job.id,
&Uuid::nil(),
flow_job.workspace_id.as_str(),
true,
serde_json::from_str("\"stopped early\"").unwrap(),
true,
same_worker_tx,
worker_dir,
Some(true),
rsmq,
worker_name,
)
.await;
job_completed_tx
.send(SendResult::UpdateFlow {
flow: flow_job.id,
success: true,
result: serde_json::from_str("\"stopped early\"").unwrap(),
stop_early_override: Some(true),
w_id: flow_job.workspace_id.clone(),
worker_dir: worker_dir.to_string(),
token: client.token.clone(),
})
.await
.map_err(|e| {
Error::InternalErr(format!(
"error sending update flow message to job completed channel: {e}"
))
})?;
// return update_flow_status_after_job_completion(
// db,
// client,
// flow_job.id,
// &Uuid::nil(),
// flow_job.workspace_id.as_str(),
// true,
// serde_json::from_str("\"stopped early\"").unwrap(),
// true,
// same_worker_tx,
// worker_dir,
// Some(true),
// rsmq,
// worker_name,
// )
// .await;
return Ok(());
}
}
}
@@ -1436,22 +1504,39 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
.await?;
if flow_job.is_flow_step {
if let Some(parent_job) = flow_job.parent_job {
update_flow_status_after_job_completion(
db,
client,
parent_job,
&flow_job.id,
&flow_job.workspace_id,
true,
&to_raw_value(&result),
false,
same_worker_tx.clone(),
&worker_dir,
None,
rsmq,
worker_name,
)
.await?;
job_completed_tx
.send(SendResult::UpdateFlow {
flow: parent_job,
success: true,
result: to_raw_value(&result),
stop_early_override: Some(true),
w_id: flow_job.workspace_id.clone(),
worker_dir: worker_dir.to_string(),
token: client.token.clone(),
})
.await
.map_err(|e| {
Error::InternalErr(format!(
"error sending update flow message to job completed channel: {e}"
))
})?;
// update_flow_status_after_job_completion(
// db,
// client,
// parent_job,
// &flow_job.id,
// &flow_job.workspace_id,
// true,
// &to_raw_value(&result),
// false,
// same_worker_tx.clone(),
// &worker_dir,
// None,
// rsmq,
// worker_name,
// )
// .await?;
return Ok(());
}
}
return Ok(());
@@ -1689,6 +1774,7 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
_ => Ok(flow_job_args),
}
};
tracing::debug!(id = %flow_job.id, root_id = %job_root, "flow job args computed");
let next_flow_transform = compute_next_flow_transform(
arc_flow_job_args.clone(),
@@ -1889,6 +1975,8 @@ async fn push_next_flow_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
)
.await?;
tracing::debug!(id = %flow_job.id, root_id = %job_root, "pushed next flow job: {uuid}");
if let FlowModuleValue::ForloopFlow { parallelism: Some(p), .. } = &module.value {
if i as u16 >= *p {
sqlx::query!(
@@ -2211,6 +2299,8 @@ async fn compute_next_flow_transform(
))
};
let delete_after_use = module.delete_after_use.unwrap_or(false);
tracing::debug!(id = %flow_job.id, "computing next flow transform for {:?}", &module.value);
match &module.value {
FlowModuleValue::Identity => trivial_next_job(JobPayload::Identity),
FlowModuleValue::Flow { path, .. } => {
frontend/src/routes/(root)/(logged)/user/(user)/workspaces/+page.svelte
+2 -1
View File
@@ -136,12 +136,13 @@
workspace.
</p>
{/if}
{#each nonAdminWorkspaces as workspace}
{#each nonAdminWorkspaces as workspace (workspace.id)}
<label class="block pb-2">
<button
class="block w-full mx-auto py-1 px-2 rounded-md border
shadow-sm text-sm font-normal mt-1 hover:ring-1 hover:ring-indigo-300"
on:click={async () => {
workspaceStore.set(undefined)
workspaceStore.set(workspace.id)
loading = true
await goto(rd ?? '/')