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

Fix build for Windows (#3235)

Browse Source 
* Fix build for Windows

* Fix typo
This commit is contained in:
Mikhail Zagurskiy
2024-02-17 10:45:38 +01:00
committed by GitHub
parent 7c65ca76c9
commit a354ebd6ea
3 changed files with 138 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-common/src/lib.rs
+13 -1
View File
@@ -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(())
backend/windmill-worker/src/common.rs
+27 -8
View File
@@ -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;
backend/windmill-worker/src/worker.rs
+98 -42
View File
@@ -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<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
let mut started = false;
#[cfg(feature = "benchmark")]
let mut infos = BenchmarkInfo { iters: 0, timings: vec![] };
let mut infos = BenchmarkInfo {
iters: 0,
timings: vec![],
};
let vacuum_shift = rand::thread_rng().gen_range(0..VACUUM_PERIOD);
@@ -1211,7 +1220,10 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
} else {
is_flow_worker = false;
if let Some((path, sender, handle)) = spawn_dedicated_worker(
SpawnWorker::Script { path: _wp.path.clone(), hash: None },
SpawnWorker::Script {
path: _wp.path.clone(),
hash: None,
},
&_wp.workspace_id,
killpill_tx.clone(),
&killpill_rx,
@@ -1711,7 +1723,10 @@ async fn spawn_dedicated_workers_for_flow(
workers.push((module.id.clone(), sender.clone(), None));
} else {
if let Some(dedi_w) = spawn_dedicated_worker(
SpawnWorker::Script { path: path.to_string(), hash: hash.clone() },
SpawnWorker::Script {
path: path.to_string(),
hash: hash.clone(),
},
w_id,
killpill_tx.clone(),
killpill_rx,
@@ -1785,7 +1800,13 @@ async fn spawn_dedicated_workers_for_flow(
workers.extend(w);
}
}
FlowModuleValue::RawScript { content, lock, path: spath, language, .. } => {
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<ScriptHash> },
RawScript { path: String, content: String, lock: Option<String>, lang: ScriptLang },
Script {
path: String,
hash: Option<ScriptHash>,
},
RawScript {
path: String,
content: String,
lock: Option<String>,
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<Box<RawValue>> {
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<R: rsmq_async::RsmqConnection + Send + Sync
&job.created_by,
&db,
&job.workspace_id,
DeployedObject::Flow { path: job_path, parent_path },
DeployedObject::Flow {
path: job_path,
parent_path,
},
deployment_message,
rsmq.clone(),
false,
@@ -3503,7 +3549,10 @@ async fn lock_modules(
.await?;
nbranches.push(b)
}
e.value = FlowModuleValue::BranchAll { branches: nbranches, parallel }
e.value = FlowModuleValue::BranchAll {
branches: nbranches,
parallel,
}
}
FlowModuleValue::BranchOne { branches, default } => {
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<R: rsmq_async::RsmqConnection + Send + Sync +
&job.created_by,
&db,
&job.workspace_id,
DeployedObject::App { path: job_path, version: id, parent_path },
DeployedObject::App {
path: job_path,
version: id,
parent_path,
},
deployment_message,
rsmq.clone(),
false,