From 88e3648ee413286769b72acc02a4af6173fa6bac Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Fri, 10 Nov 2023 13:17:39 +0100 Subject: [PATCH] fix: optimize single step iterative forloops (#2596) * cc * fixed --- backend/Cargo.lock | 198 +++++++++++------- backend/windmill-worker/src/worker_flow.rs | 194 +++++++++++------ .../flows/content/FlowModuleComponent.svelte | 5 +- python-client/wmill/wmill/client.py | 14 +- 4 files changed, 266 insertions(+), 145 deletions(-) diff --git a/backend/Cargo.lock b/backend/Cargo.lock index b6e20318bf..99c3911daf 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -56,7 +56,7 @@ version = "0.7.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5a824f2aa7e75a0c98c5a504fceb80649e9c35265d44525b5f94de4771a395cd" dependencies = [ - "getrandom 0.2.11", + "getrandom 0.2.10", "once_cell", "version_check", ] @@ -68,7 +68,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "91429305e9f0a25f6205c5b8e0d2db09e0708a7a6df0f42212bb56c32c8ac97a" dependencies = [ "cfg-if", - "getrandom 0.2.11", + "getrandom 0.2.10", "once_cell", "version_check", "zerocopy", @@ -247,8 +247,8 @@ dependencies = [ "pin-project-lite", "tokio", "xz2", - "zstd 0.11.2+zstd.1.5.2", - "zstd-safe 5.0.2+zstd.1.5.2", + "zstd", + "zstd-safe", ] [[package]] @@ -827,15 +827,6 @@ dependencies = [ "alloc-stdlib", ] -[[package]] -name = "btoi" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9dd6407f73a9b8b6162d8a2ef999fe6afd7cc15902ebf42c5cd296addf17e0ad" -dependencies = [ - "num-traits", -] - [[package]] name = "built" version = "0.6.1" @@ -2574,9 +2565,9 @@ dependencies = [ [[package]] name = "getrandom" -version = "0.2.11" +version = "0.2.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fe9006bed769170c11f845cf00c7c1e9092aeb3f268e007c3e760ac68008070f" +checksum = "be4136b2a15dd319360be1c07d9933517ccf0be8f16bf62a3bee4f0d618df427" dependencies = [ "cfg-if", "js-sys", @@ -3129,15 +3120,6 @@ dependencies = [ "simple_asn1", ] -[[package]] -name = "keyed_priority_queue" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2d63b6407b66fc81fc539dccf3ddecb669f393c5101b6a2be3976c95099a06e8" -dependencies = [ - "indexmap 1.9.3", -] - [[package]] name = "konst" version = "0.2.19" @@ -3206,6 +3188,79 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" +[[package]] +name = "lexical" +version = "6.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7aefb36fd43fef7003334742cbf77b243fcd36418a1d1bdd480d613a67968f6" +dependencies = [ + "lexical-core", +] + +[[package]] +name = "lexical-core" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2cde5de06e8d4c2faabc400238f9ae1c74d5412d03a7bd067645ccbc47070e46" +dependencies = [ + "lexical-parse-float", + "lexical-parse-integer", + "lexical-util", + "lexical-write-float", + "lexical-write-integer", +] + +[[package]] +name = "lexical-parse-float" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "683b3a5ebd0130b8fb52ba0bdc718cc56815b6a097e28ae5a6997d0ad17dc05f" +dependencies = [ + "lexical-parse-integer", + "lexical-util", + "static_assertions", +] + +[[package]] +name = "lexical-parse-integer" +version = "0.8.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d0994485ed0c312f6d965766754ea177d07f9c00c9b82a5ee62ed5b47945ee9" +dependencies = [ + "lexical-util", + "static_assertions", +] + +[[package]] +name = "lexical-util" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5255b9ff16ff898710eb9eb63cb39248ea8a5bb036bea8085b1a767ff6c4e3fc" +dependencies = [ + "static_assertions", +] + +[[package]] +name = "lexical-write-float" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "accabaa1c4581f05a3923d1b4cfd124c329352288b7b9da09e766b0668116862" +dependencies = [ + "lexical-util", + "lexical-write-integer", + "static_assertions", +] + +[[package]] +name = "lexical-write-integer" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1b6f3d1f4422866b68192d62f77bc5c700bee84f3069f2469d7bc8c77852446" +dependencies = [ + "lexical-util", + "static_assertions", +] + [[package]] name = "libc" version = "0.2.150" @@ -3303,9 +3358,9 @@ checksum = "ef53942eb7bf7ff43a617b3e2c1c4a5ecf5944a7c1bc12d7ee39bbb15e5c1519" [[package]] name = "linux-raw-sys" -version = "0.4.11" +version = "0.4.10" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "969488b55f8ac402214f3f5fd243ebb7206cf82de60d3172994707a4bcc2b829" +checksum = "da2479e8c062e40bf0066ffa0bc823de0a9368974af99c9f6df941d2c231e03f" [[package]] name = "lock_api" @@ -3338,11 +3393,11 @@ dependencies = [ [[package]] name = "lru" -version = "0.12.0" +version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1efa59af2ddfad1854ae27d75009d538d0998b4b2fd47083e743ac1a10e46c60" +checksum = "718e8fae447df0c7e1ba7f5189829e63fd536945c8988d61444c19039f16b670" dependencies = [ - "hashbrown 0.14.2", + "hashbrown 0.13.2", ] [[package]] @@ -3621,9 +3676,9 @@ dependencies = [ [[package]] name = "mysql_async" -version = "0.33.0" +version = "0.32.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6750b17ce50f8f112ef1a8394121090d47c596b56a6a17569ca680a9626e2ef2" +checksum = "f5272f59b5b1f93d65f7f826c1f025d6e410e89fb50a67e05aa20b35a55a8c0a" dependencies = [ "bytes", "crossbeam", @@ -3631,17 +3686,16 @@ dependencies = [ "futures-core", "futures-sink", "futures-util", - "keyed_priority_queue", "lazy_static", "lru", "mio", "mysql_common", "native-tls", "once_cell", - "pem 3.0.2", + "pem 2.0.1", "percent-encoding", "pin-project", - "rand 0.8.5", + "priority-queue", "serde", "serde_json", "socket2 0.5.5", @@ -3655,16 +3709,15 @@ dependencies = [ [[package]] name = "mysql_common" -version = "0.31.0" +version = "0.30.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "06f19e4cfa0ab5a76b627cec2d81331c49b034988eaf302c3bafeada684eadef" +checksum = "57349d5a326b437989b6ee4dc8f2f34b0cc131202748414712a8e7d98952fc8c" dependencies = [ "base64 0.21.5", - "bigdecimal 0.4.2", + "bigdecimal 0.3.1", "bindgen 0.69.1", "bitflags 2.4.1", "bitvec", - "btoi", "byteorder", "bytes", "cc", @@ -3673,6 +3726,7 @@ dependencies = [ "flate2", "frunk", "lazy_static", + "lexical", "mysql-common-derive", "num-bigint", "num-traits", @@ -3689,7 +3743,6 @@ dependencies = [ "thiserror", "time", "uuid 1.5.0", - "zstd 0.12.4", ] [[package]] @@ -4073,6 +4126,16 @@ dependencies = [ "base64 0.13.1", ] +[[package]] +name = "pem" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b13fe415cdf3c8e44518e18a7c95a13431d9bdf6d15367d82b23c377fdd441a" +dependencies = [ + "base64 0.21.5", + "serde", +] + [[package]] name = "pem" version = "3.0.2" @@ -4463,6 +4526,16 @@ dependencies = [ "syn 2.0.39", ] +[[package]] +name = "priority-queue" +version = "1.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fff39edfcaec0d64e8d0da38564fad195d2d51b680940295fcc307366e101e61" +dependencies = [ + "autocfg", + "indexmap 1.9.3", +] + [[package]] name = "proc-macro-crate" version = "0.1.5" @@ -4776,7 +4849,7 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ - "getrandom 0.2.11", + "getrandom 0.2.10", ] [[package]] @@ -4899,7 +4972,7 @@ version = "0.4.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a18479200779601e498ada4e8c1e1f50e3ee19deb0259c25825a98b5603b2cb4" dependencies = [ - "getrandom 0.2.11", + "getrandom 0.2.10", "libredox", "thiserror", ] @@ -5048,7 +5121,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fb0205304757e5d899b9c2e448b867ffd03ae7f988002e47cd24954391394d0b" dependencies = [ "cc", - "getrandom 0.2.11", + "getrandom 0.2.10", "libc", "spin 0.9.8", "untrusted 0.9.0", @@ -5256,7 +5329,7 @@ dependencies = [ "bitflags 2.4.1", "errno", "libc", - "linux-raw-sys 0.4.11", + "linux-raw-sys 0.4.10", "windows-sys 0.48.0", ] @@ -5298,9 +5371,9 @@ dependencies = [ [[package]] name = "rustls-pemfile" -version = "1.0.4" +version = "1.0.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1c74cae0a4cf6ccbbf5f359f08efdf8ee7e1dc532573bf0db71968cb56b1448c" +checksum = "2d3987094b1d07b653b7dfdc3f70ce9a1da9c51ac18c1b06b662e4f9a0e9f4b2" dependencies = [ "base64 0.21.5", ] @@ -5851,9 +5924,9 @@ dependencies = [ [[package]] name = "smallvec" -version = "1.11.2" +version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4dccd0940a2dcdf68d092b8cbab7dc0ad8fa938bf95787e1b916b0e3d0e8e970" +checksum = "942b4a808e05215192e39f4ab80813e599068285906cc91aa64f923db842bd5a" [[package]] name = "smart-default" @@ -7018,7 +7091,7 @@ dependencies = [ "clap", "derive_builder", "esaxx-rs", - "getrandom 0.2.11", + "getrandom 0.2.10", "indicatif", "itertools 0.11.0", "lazy_static", @@ -7767,7 +7840,7 @@ version = "0.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bc5cf98d8186244414c848017f0e2676b3fcb46807f6668a97dfe67359a3c4b7" dependencies = [ - "getrandom 0.2.11", + "getrandom 0.2.10", ] [[package]] @@ -7776,7 +7849,7 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "88ad59a7560b41a70d191093a945f0b87bc1deeda46fb237479708a1d6b6cdfc" dependencies = [ - "getrandom 0.2.11", + "getrandom 0.2.10", "serde", ] @@ -8336,7 +8409,7 @@ name = "windmill-parser-wasm" version = "1.204.0" dependencies = [ "anyhow", - "getrandom 0.2.11", + "getrandom 0.2.10", "serde_json", "wasm-bindgen", "wasm-bindgen-test", @@ -8717,7 +8790,7 @@ dependencies = [ "pbkdf2", "sha1", "time", - "zstd 0.11.2+zstd.1.5.2", + "zstd", ] [[package]] @@ -8726,16 +8799,7 @@ version = "0.11.2+zstd.1.5.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20cc960326ece64f010d2d2107537f26dc589a6573a316bd5b1dba685fa5fde4" dependencies = [ - "zstd-safe 5.0.2+zstd.1.5.2", -] - -[[package]] -name = "zstd" -version = "0.12.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1a27595e173641171fc74a1232b7b1c7a7cb6e18222c11e9dfb9888fa424c53c" -dependencies = [ - "zstd-safe 6.0.6", + "zstd-safe", ] [[package]] @@ -8748,16 +8812,6 @@ dependencies = [ "zstd-sys", ] -[[package]] -name = "zstd-safe" -version = "6.0.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee98ffd0b48ee95e6c5168188e44a54550b1564d9d530ee21d5f0eaed1069581" -dependencies = [ - "libc", - "zstd-sys", -] - [[package]] name = "zstd-sys" version = "2.0.9+zstd.1.5.5" diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index 900dceaac8..b5051ecf51 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -1706,22 +1706,35 @@ async fn push_next_flow_job iterator: None, .. } => args.as_ref().map(|args| args.clone()), - NextStatus::NextLoopIteration(NextIteration { new_args, .. }) => { + NextStatus::NextLoopIteration { + next: NextIteration { new_args, .. }, + simple_input_transforms, + } => { let mut args = if let Ok(args) = args.as_ref() { args.clone() } else { HashMap::new() }; insert_iter_arg(&mut args, "iter".to_string(), to_raw_value(new_args)); - Ok(args) - // tracing::error!( - // "{a:?} {new_args:?} {:?}", - // to_raw_value(&MergeArgs { a: HashMap::new(), b: new_args.to_owned() }) - // ); - // Ok(to_raw_value(&MergeArgs { - // a: HashMap::new(), - // b: new_args.to_owned(), - // })) + + args.insert("iter".to_string(), to_raw_value(new_args)); + if let Some(input_transforms) = simple_input_transforms { + let ctx = get_transform_context(&flow_job, &previous_id, &status).await?; + transform_inp = transform_input( + Arc::new(args), + arc_last_job_result.clone(), + input_transforms, + resumes.clone(), + resume.clone(), + approvers.clone(), + &ctx, + client, + ) + .await; + transform_inp.as_ref().map(|args| args.clone()) + } else { + Ok(args) + } } NextStatus::AllFlowJobs { branchall: None, @@ -1832,7 +1845,10 @@ async fn push_next_flow_job }; let first_uuid = uuids[0]; let new_status = match next_status { - NextStatus::NextLoopIteration(NextIteration { index, itered, mut flow_jobs, .. }) => { + NextStatus::NextLoopIteration { + next: NextIteration { index, itered, mut flow_jobs, .. }, + .. + } => { let uuid = one_uuid?; flow_jobs.push(uuid); @@ -2028,7 +2044,10 @@ enum NextStatus { NextStep, BranchChosen(BranchChosen), NextBranchStep(NextBranch), - NextLoopIteration(NextIteration), + NextLoopIteration { + next: NextIteration, + simple_input_transforms: Option>, + }, AllFlowJobs { branchall: Option, iterator: Option, @@ -2154,7 +2173,9 @@ async fn compute_next_flow_transform( && modules[0].sleep.is_none() && modules[0].suspend.is_none() && modules[0].cache_ttl.is_none() - && flow.failure_module.is_none(); + && (modules[0].mock.is_none() + && modules[0].mock.as_ref().is_some_and(|m| !m.enabled) + && flow.failure_module.is_none()); let next_loop_status = match status_module { FlowStatusModule::WaitingForPriorSteps { .. } @@ -2280,63 +2301,71 @@ async fn compute_next_flow_transform( } let modules = (*modules).clone(); let inner_path = Some(format!("{}/loop-{}", flow_job.script_path(), ns.index)); - let continue_payload = ContinuePayload::SingleJob(JobPayloadWithTag { - payload: JobPayload::RawFlow { - value: FlowValue { - modules, - failure_module: fm, - same_worker: flow.same_worker, - concurrent_limit: None, - concurrency_time_window_s: None, - skip_expr: None, - cache_ttl: None, - ws_error_handler_muted: None, - priority: None, + if is_simple { + let payload = payload_from_simple_module( + &modules[0].value, + db, + flow_job, + module, + inner_path, + ) + .await?; + Ok(NextFlowTransform::Continue( + ContinuePayload::SingleJob(payload), + NextStatus::NextLoopIteration { + next: ns, + simple_input_transforms: if is_simple { + match &modules[0].value { + FlowModuleValue::Script { input_transforms, .. } + | FlowModuleValue::RawScript { input_transforms, .. } + | FlowModuleValue::Flow { input_transforms, .. } => { + Some(input_transforms.clone()) + } + _ => None, + } + } else { + None + }, }, - path: inner_path, - restarted_from: None, - }, - tag: None, - }); - Ok(NextFlowTransform::Continue( - continue_payload, - NextStatus::NextLoopIteration(ns), - )) + )) + } else { + Ok(NextFlowTransform::Continue( + ContinuePayload::SingleJob(JobPayloadWithTag { + payload: JobPayload::RawFlow { + value: FlowValue { + modules, + failure_module: fm, + same_worker: flow.same_worker, + concurrent_limit: None, + concurrency_time_window_s: None, + skip_expr: None, + cache_ttl: None, + ws_error_handler_muted: None, + priority: None, + }, + path: inner_path, + restarted_from: None, + }, + tag: None, + }), + NextStatus::NextLoopIteration { + next: ns, + simple_input_transforms: None, + }, + )) + } } LoopStatus::ParallelIteration { itered, .. } => { let inner_path = Some(format!("{}/loop-parrallel", flow_job.script_path(),)); let continue_payload = if is_simple { - let payload = match &modules[0].value { - FlowModuleValue::Flow { path, .. } => flow_to_payload(path), - FlowModuleValue::Script { - path: script_path, - hash: script_hash, - .. - } => { - script_to_payload(script_hash, script_path, db, flow_job, module) - .await? - } - FlowModuleValue::RawScript { - path, - content, - language, - lock, - tag, - concurrent_limit, - concurrency_time_window_s, - .. - } => raw_script_to_payload( - path.clone().or(inner_path), - content, - language, - lock, - concurrent_limit, - concurrency_time_window_s, - module, - tag, - ), - _ => unreachable!("is simple flow"), - }; + let payload = payload_from_simple_module( + &modules[0].value, + db, + flow_job, + module, + inner_path, + ) + .await?; ContinuePayload::ForloopJobs { n: itered.len(), payload: payload } } else { let payload = { @@ -2573,6 +2602,41 @@ async fn compute_next_flow_transform( } } +async fn payload_from_simple_module( + value: &FlowModuleValue, + db: &sqlx::Pool, + flow_job: &QueuedJob, + module: &FlowModule, + inner_path: Option, +) -> Result { + Ok(match value { + FlowModuleValue::Flow { path, .. } => flow_to_payload(path), + FlowModuleValue::Script { path: script_path, hash: script_hash, .. } => { + script_to_payload(script_hash, script_path, db, flow_job, module).await? + } + FlowModuleValue::RawScript { + path, + content, + language, + lock, + tag, + concurrent_limit, + concurrency_time_window_s, + .. + } => raw_script_to_payload( + path.clone().or(inner_path), + content, + language, + lock, + concurrent_limit, + concurrency_time_window_s, + module, + tag, + ), + _ => unreachable!("is simple flow"), + }) +} + fn raw_script_to_payload( path: Option, content: &String, diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index d2dca529eb..d16623cf8d 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -119,12 +119,12 @@ modulePreview?.runTestWithStepArgs() } } - + let inputTransformSchemaForm: InputTransformSchemaForm | undefined = undefined async function reload(flowModule: FlowModule) { try { const { input_transforms, schema } = await loadSchemaFromModule(flowModule) validCode = true - + inputTransformSchemaForm?.setArgs(input_transforms) if ( flowModule.value.type == 'rawscript' || flowModule.value.type == 'script' || @@ -314,6 +314,7 @@ error={failureModule} > An workspace=get_workspace(), path=path, client=create_client() ) try: - parsed = json.loads(parsed.content.decode("utf-8")) + content = parsed.content.decode("utf-8") + parsed = json.loads(content) except: parsed = None @@ -216,7 +217,7 @@ def get_resource(path: str | None = None, none_if_undefined: bool = False) -> An return None else: raise Exception( - f"Resource at path {path} does not exist or you do not have read permissions on it" + f"Resource at path {path} does not exist or you do not have read permissions on it: {content}" ) return parsed @@ -331,13 +332,14 @@ def get_variable(path: str) -> str: res = get_variable_api.sync_detailed( workspace=get_workspace(), path=path, client=create_client() - ).parsed - if res is None: + ) + parsed = res.parsed + if parsed is None: raise Exception( - f"Variable at path {path} does not exist or you do not have read permissions on it" + f"Variable at path {path} does not exist or you do not have read permissions on it: {res.content.decode('utf-8')}" ) - return res + return parsed def set_variable(path: str, value: str) -> None: