From a354ebd6ea81893cf7880d32aabfbcc7543e4a10 Mon Sep 17 00:00:00 2001 From: Mikhail Zagurskiy <70192849+mikhailzagurskiy@users.noreply.github.com> Date: Sat, 17 Feb 2024 11:45:38 +0200 Subject: [PATCH] Fix build for Windows (#3235) * Fix build for Windows * Fix typo --- backend/windmill-common/src/lib.rs | 14 ++- backend/windmill-worker/src/common.rs | 35 +++++-- backend/windmill-worker/src/worker.rs | 140 ++++++++++++++++++-------- 3 files changed, 138 insertions(+), 51 deletions(-) diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 9f5b507929..1ddf8673af 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -75,15 +75,18 @@ pub async fn shutdown_signal( mut rx: tokio::sync::broadcast::Receiver<()>, ) -> anyhow::Result<()> { use std::io; - use tokio::signal::unix::SignalKind; + #[cfg(target_os = "linux")] async fn terminate() -> io::Result<()> { + use tokio::signal::unix::SignalKind; tokio::signal::unix::signal(SignalKind::terminate())? .recv() .await; + Ok(()) } + #[cfg(target_os = "linux")] tokio::select! { _ = terminate() => {}, _ = tokio::signal::ctrl_c() => {}, @@ -91,6 +94,15 @@ pub async fn shutdown_signal( tracing::info!("shutdown monitor received killpill"); }, } + + #[cfg(not(target_os = "linux"))] + tokio::select! { + _ = tokio::signal::ctrl_c() => {}, + _ = rx.recv() => { + tracing::info!("shutdown monitor received killpill"); + }, + } + println!("signal received, starting graceful shutdown"); let _ = tx.send(()); Ok(()) diff --git a/backend/windmill-worker/src/common.rs b/backend/windmill-worker/src/common.rs index 20d3893c1f..f0d11f817e 100644 --- a/backend/windmill-worker/src/common.rs +++ b/backend/windmill-worker/src/common.rs @@ -1,7 +1,11 @@ use async_recursion::async_recursion; use itertools::Itertools; + +#[cfg(target_os = "linux")] use nix::sys::signal::{self, Signal}; +#[cfg(target_os = "linux")] use nix::unistd::Pid; + use regex::Regex; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; @@ -23,13 +27,14 @@ use windmill_common::{ use anyhow::Result; use windmill_queue::CanceledBy; +#[cfg(target_os = "linux")] +use std::os::unix::process::ExitStatusExt; + use std::{ borrow::Borrow, collections::{hash_map::DefaultHasher, HashMap}, hash::{Hash, Hasher}, - io, - os::unix::process::ExitStatusExt, - panic, + io, panic, time::Duration, }; @@ -609,7 +614,9 @@ pub async fn handle_child( if let Some(id) = child.id() { if *MAX_WAIT_FOR_SIGINT > 0 { + #[cfg(target_os = "linux")] signal::kill(Pid::from_raw(id as i32), Signal::SIGINT).unwrap(); + for _ in 0..*MAX_WAIT_FOR_SIGINT { if child.try_wait().is_ok_and(|x| x.is_some()) { break; @@ -622,7 +629,9 @@ pub async fn handle_child( } } if sigterm { + #[cfg(target_os = "linux")] signal::kill(Pid::from_raw(id as i32), Signal::SIGTERM).unwrap(); + for _ in 0..*MAX_WAIT_FOR_SIGTERM { if child.try_wait().is_ok_and(|x| x.is_some()) { break; @@ -758,12 +767,18 @@ pub async fn handle_child( } else if let Some(code) = status.code() { Err(error::Error::ExitStatus(code)) } else { - Err(error::Error::ExecutionErr(format!( + #[cfg(target_os = "linux")] + return Err(error::Error::ExecutionErr(format!( "process terminated by signal: {:#?}, stopped_signal: {:#?}, core_dumped: {}", status.signal(), status.stopped_signal(), status.core_dumped() - ))) + ))); + + #[cfg(not(target_os = "linux"))] + return Err(error::Error::ExecutionErr(String::from( + "process terminated by signal", + ))); } } Ok(Err(kill_reason)) => Err(Error::ExecutionErr(format!( @@ -1063,9 +1078,13 @@ pub async fn get_cached_resource_value_if_valid( } for (s3_file_key, s3_file_etag) in s3_etags { if let Some(object_store_resource) = object_store_resource_opt.clone() { - let etag = - get_etag_or_empty(&object_store_resource, S3Object { s3: s3_file_key.clone() }) - .await; + let etag = get_etag_or_empty( + &object_store_resource, + S3Object { + s3: s3_file_key.clone(), + }, + ) + .await; if etag.is_none() || etag.clone().unwrap() != s3_file_etag { tracing::warn!("S3 file etag for '{}' has changed. Value from cache is {:?} while current value from S3 is {:?}. Cache will be invalidated", s3_file_key.clone(), s3_file_etag, etag); return None; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index a7420fedbe..86a2e41a01 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -51,8 +51,14 @@ use windmill_queue::{ use serde_json::{json, value::RawValue, Value}; +#[cfg(target_os = "linux")] +use tokio::fs::symlink; + +#[cfg(target_os = "windows")] +use tokio::fs::symlink_file as symlink; + use tokio::{ - fs::{symlink, DirBuilder}, + fs::DirBuilder, sync::{ mpsc::{self, Sender}, Barrier, RwLock, @@ -1142,7 +1148,10 @@ pub async fn run_worker { + FlowModuleValue::RawScript { + content, + lock, + path: spath, + language, + .. + } => { if let Some(dedi_w) = spawn_dedicated_worker( SpawnWorker::RawScript { path: spath.clone().unwrap_or(path.to_string()), @@ -1816,8 +1837,16 @@ async fn spawn_dedicated_workers_for_flow( } enum SpawnWorker { - Script { path: String, hash: Option }, - RawScript { path: String, content: String, lock: Option, lang: ScriptLang }, + Script { + path: String, + hash: Option, + }, + RawScript { + path: String, + content: String, + lock: Option, + lang: ScriptLang, + }, } // spawn one dedicated worker and return the key, the channel sender and the join handle @@ -1898,7 +1927,12 @@ async fn spawn_dedicated_worker( return None; } } - SpawnWorker::RawScript { content, lock, lang, .. } => (content, lock, Some(lang), None), + SpawnWorker::RawScript { + content, + lock, + lang, + .. + } => (content, lock, Some(lang), None), }; match language { @@ -2809,7 +2843,12 @@ async fn get_script_content_by_hash( .fetch_optional(db) .await? .ok_or_else(|| Error::InternalErr(format!("expected content and lock")))?; - Ok(ContentReqLangEnvs { content: r.0, lockfile: r.1, language: r.2, envs: r.3 }) + Ok(ContentReqLangEnvs { + content: r.0, + lockfile: r.1, + language: r.2, + envs: r.3, + }) } #[tracing::instrument(level = "trace", skip_all)] @@ -2825,35 +2864,39 @@ async fn handle_code_execution_job( base_internal_url: &str, worker_name: &str, ) -> error::Result> { - let ContentReqLangEnvs { content: inner_content, lockfile: requirements_o, language, envs } = - match job.job_kind { - JobKind::Preview => ContentReqLangEnvs { - content: job - .raw_code - .clone() - .unwrap_or_else(|| "no raw code".to_owned()), - lockfile: job.raw_lock.clone(), - language: job.language.to_owned(), - envs: None, - }, - JobKind::Script_Hub => { - get_hub_script_content_and_requirements(job.script_path.clone(), db).await? - } - JobKind::Script => { - get_script_content_by_hash( - &job.script_hash.unwrap_or(ScriptHash(0)), - &job.workspace_id, - db, - ) - .await? - } - JobKind::DeploymentCallback => { - get_script_content_by_path(job.script_path.clone(), &job.workspace_id, db).await? - } - _ => unreachable!( - "handle_code_execution_job should never be reachable with a non-code execution job" - ), - }; + let ContentReqLangEnvs { + content: inner_content, + lockfile: requirements_o, + language, + envs, + } = match job.job_kind { + JobKind::Preview => ContentReqLangEnvs { + content: job + .raw_code + .clone() + .unwrap_or_else(|| "no raw code".to_owned()), + lockfile: job.raw_lock.clone(), + language: job.language.to_owned(), + envs: None, + }, + JobKind::Script_Hub => { + get_hub_script_content_and_requirements(job.script_path.clone(), db).await? + } + JobKind::Script => { + get_script_content_by_hash( + &job.script_hash.unwrap_or(ScriptHash(0)), + &job.workspace_id, + db, + ) + .await? + } + JobKind::DeploymentCallback => { + get_script_content_by_path(job.script_path.clone(), &job.workspace_id, db).await? + } + _ => unreachable!( + "handle_code_execution_job should never be reachable with a non-code execution job" + ), + }; if language == Some(ScriptLang::Postgresql) { return do_postgresql(job, &client, &inner_content, db).await; @@ -3387,7 +3430,10 @@ async fn handle_flow_dependency_job { let mut nbranches = vec![]; @@ -3540,7 +3589,10 @@ async fn lock_modules( token, ) .await?; - e.value = FlowModuleValue::BranchOne { branches: nbranches, default }; + e.value = FlowModuleValue::BranchOne { + branches: nbranches, + default, + }; } _ => (), }; @@ -3801,7 +3853,11 @@ async fn handle_app_dependency_job