feat: better locking (#3412)

* earlyBackendStop * fix(frontend): Flow step input description (#3409) * fix(frontend): Add missing descrpition in a flow step input * fix(frontend): Add missing descrpition in a flow step input * fix(frontend): Fix color picker layout (#3411) * feat: add bit support in pg (#3407) * fix sqlx * fix: remove dependency on semver intersect for bun * sqlx fix * improve early kill behavior --------- Co-authored-by: Faton Ramadani <faton.ramadani14@gmail.com> Co-authored-by: HugoCasa <hugo@casademont.ch>
2026-08-18 16:02:10 +00:00 · 2024-03-14 13:57:54 +01:00
parent 5367e1f5d8
commit 72bf49b757
4 changed files with 69 additions and 69 deletions
@@ -1,26 +1,24 @@
-FROM python:3.11-slim-buster as nsjail
+ARG DEBIAN_IMAGE=debian:bookworm-slim
+ARG RUST_IMAGE=rust:1.75-slim-bookworm
+ARG PYTHON_IMAGE=python:3.11.4-slim-bookworm

-WORKDIR /nsjail
+FROM ${DEBIAN_IMAGE} as downloader

-RUN apt-get -y update \
-    && apt-get install -y \
-    bison=2:3.3.* \
-    flex=2.6.* \
-    g++=4:8.3.* \
-    gcc=4:8.3.* \
-    git=1:2.20.* \
-    libprotobuf-dev=3.6.* \
-    libnl-route-3-dev=3.4.* \
-    make=4.2.* \
-    pkg-config=0.29-6 \
-    protobuf-compiler=3.6.*
+ARG TARGETPLATFORM

-RUN git clone -b master --single-branch https://github.com/google/nsjail.git . \
-    && git checkout dccf911fd2659e7b08ce9507c25b2b38ec2c5800
-RUN make
+SHELL ["/bin/bash", "-c"]
+
+RUN apt update -y
+RUN apt install -y unzip curl
+
+RUN [ "$TARGETPLATFORM" == "linux/amd64" ] && curl -Lsf https://github.com/denoland/deno/releases/download/v1.41.0/deno-x86_64-unknown-linux-gnu.zip -o deno.zip || true
+RUN [ "$TARGETPLATFORM" == "linux/arm64" ] && curl -Lsf https://github.com/denoland/deno/releases/download/v1.41.0/deno-aarch64-unknown-linux-gnu.zip -o deno.zip || true


-FROM rust:slim-buster as builder
+RUN unzip deno.zip && rm deno.zip
+
+
+FROM ${RUST_IMAGE} as builder

 RUN apt-get update && apt-get install -y git libssl-dev pkg-config

@@ -36,8 +34,8 @@ RUN apt-get update \
    && apt-get install -y ca-certificates tzdata libpq5 cmake\
    make build-essential libssl-dev zlib1g-dev libbz2-dev libreadline-dev \
    libsqlite3-dev wget curl llvm libncurses5-dev libncursesw5-dev xz-utils tk-dev libxml2-dev \
-    libxmlsec1-dev libffi-dev liblzma-dev mecab-ipadic-utf8 libgdbm-dev libc6-dev git libprotobuf-dev=3.6.* libnl-route-3-dev=3.4.* \
-    libv8-dev tesseract-ocr nodejs npm clang libclang-dev\
+    libxmlsec1-dev libffi-dev liblzma-dev mecab-ipadic-utf8 libgdbm-dev libc6-dev git libprotobuf-dev libnl-route-3-dev \
+    libv8-dev  nodejs npm clang libclang-dev\
    && rm -rf /var/lib/apt/lists/*


@@ -47,19 +45,21 @@ ENV GO_PATH=/usr/local/go/bin/go

 ENV TZ=Etc/UTC

-ENV PYTHON_VERSION 3.10.4
+ENV PYTHON_VERSION 3.11.4

 RUN wget https://www.python.org/ftp/python/${PYTHON_VERSION}/Python-${PYTHON_VERSION}.tgz \
    && tar -xf Python-${PYTHON_VERSION}.tgz && cd Python-${PYTHON_VERSION}/ && ./configure --enable-optimizations \
    && make -j 4 && make install

 RUN /usr/local/bin/python3 -m pip install pip-tools
-RUN /usr/local/bin/python3 -m pip install nltk
-RUN mkdir -p /nsjail_data/python && HOME=/nsjail_data/python /usr/local/bin/python3 -m nltk.downloader vader_lexicon

-COPY --from=nsjail /nsjail/nsjail /bin/nsjail
-COPY --from=oven/bun:1.0 /usr/local/bin/bun /usr/bin/bun
-COPY --from=denoland/deno:1.40.5 /usr/bin/deno /usr/bin/deno
+COPY --from=oven/bun:1.0.29 /usr/local/bin/bun /usr/bin/bun
+
+
+RUN [ "$TARGETPLATFORM" == "linux/amd64" ] && curl -Lsf https://github.com/denoland/deno/releases/download/v1.41.0/deno-x86_64-unknown-linux-gnu.zip -o deno.zip || true
+RUN [ "$TARGETPLATFORM" == "linux/arm64" ] && curl -Lsf https://github.com/denoland/deno/releases/download/v1.41.0/deno-aarch64-unknown-linux-gnu.zip -o deno.zip || true
+
+COPY --from=downloader --chmod=755 /deno /usr/bin/deno

 RUN apt-get update \
    && apt-get install -y postgresql-client --allow-unauthenticated
@@ -12,13 +12,9 @@ use rand::Rng;
 use sqlx::{postgres::PgListener, Pool, Postgres};
 use std::{
    net::{IpAddr, Ipv4Addr, SocketAddr},
-    sync::Arc,
    time::Duration,
 };
-use tokio::{
-    fs::{metadata, DirBuilder},
-    sync::RwLock,
-};
+use tokio::fs::{metadata, DirBuilder};
 use windmill_api::HTTP_CLIENT;
 use windmill_common::{
    global_settings::{
@@ -253,10 +249,13 @@ async fn windmill_main() -> anyhow::Result<()> {
        // migration code to avoid break
        windmill_api::migrate_db(&db).await?;
    }
-    let (killpill_tx, killpill_rx) = tokio::sync::broadcast::channel::<()>(2);
-    let (killpill_phase2_tx, killpill_phase2_rx) = tokio::sync::broadcast::channel::<()>(2);
+    let (killpill_tx, mut killpill_rx) = tokio::sync::broadcast::channel::<()>(2);
+    let mut monitor_killpill_rx = killpill_tx.subscribe();
+    let server_killpill_rx = killpill_tx.subscribe();
+    let (killpill_phase2_tx, _killpill_phase2_rx) = tokio::sync::broadcast::channel::<()>(2);
+
    let shutdown_signal =
-        windmill_common::shutdown_signal(killpill_tx.clone(), killpill_rx.resubscribe());
+        windmill_common::shutdown_signal(killpill_tx.clone(), killpill_tx.subscribe());

    #[cfg(feature = "enterprise")]
    tracing::info!(
@@ -323,7 +322,7 @@ Windmill Community Edition {GIT_VERSION}
                    db.clone(),
                    rsmq2,
                    addr,
-                    killpill_phase2_rx.resubscribe(),
+                    server_killpill_rx,
                    base_internal_tx,
                    server_mode,
                )
@@ -339,22 +338,26 @@ Windmill Community Edition {GIT_VERSION}
        };

        let workers_f = async {
-            let base_internal_url = base_internal_rx.await?;
-            if worker_mode {
-                run_workers(
-                    db.clone(),
-                    killpill_rx.resubscribe(),
-                    killpill_tx.clone(),
-                    num_workers,
-                    base_internal_url.clone(),
-                    rsmq.clone(),
-                    mode.clone() == Mode::Agent,
-                )
-                .await?;
-                tracing::info!("All workers exited.");
-                killpill_tx.send(())?;
-            } else {
-                killpill_rx.resubscribe().recv().await?;
+            let mut rx = killpill_rx.resubscribe();
+
+            if !killpill_rx.try_recv().is_ok() {
+                let base_internal_url = base_internal_rx.await?;
+                if worker_mode {
+                    run_workers(
+                        db.clone(),
+                        rx,
+                        killpill_tx.clone(),
+                        num_workers,
+                        base_internal_url.clone(),
+                        rsmq.clone(),
+                        mode.clone() == Mode::Agent,
+                    )
+                    .await?;
+                    tracing::info!("All workers exited.");
+                    killpill_tx.send(())?;
+                } else {
+                    rx.recv().await?;
+                }
            }
            tracing::info!("Starting phase 2 of shutdown");
            killpill_phase2_tx.send(())?;
@@ -366,7 +369,6 @@ Windmill Community Edition {GIT_VERSION}
            let tx = killpill_tx.clone();
            let rsmq = rsmq.clone();

-            let mut rx = killpill_rx.resubscribe();
            let base_internal_url = base_internal_url.to_string();
            let h = tokio::spawn(async move {
                let mut listener = retry_listen_pg(&db).await;
@@ -506,7 +508,7 @@ Windmill Community Edition {GIT_VERSION}
                                }
                            };
                        },
-                        _ = rx.recv() => {
+                        _ = monitor_killpill_rx.recv() => {
                                println!("received killpill for monitor job");
                                break;
                        }
@@ -526,12 +528,8 @@ Windmill Community Edition {GIT_VERSION}
                tracing::error!("Metrics are only available in the EE, ignoring...");

                #[cfg(feature = "enterprise")]
-                windmill_common::serve_metrics(
-                    *METRICS_ADDR,
-                    killpill_phase2_rx.resubscribe(),
-                    num_workers > 0,
-                )
-                .await;
+                windmill_common::serve_metrics(*METRICS_ADDR, _killpill_phase2_rx, num_workers > 0)
+                    .await;
            }
            Ok(()) as anyhow::Result<()>
        };
@@ -610,13 +608,22 @@ fn display_config(envs: &[&str]) {

 pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 'static>(
    db: Pool<Postgres>,
-    rx: tokio::sync::broadcast::Receiver<()>,
+    mut rx: tokio::sync::broadcast::Receiver<()>,
    tx: tokio::sync::broadcast::Sender<()>,
    num_workers: i32,
    base_internal_url: String,
    rsmq: Option<R>,
    agent_mode: bool,
 ) -> anyhow::Result<()> {
+    let mut killpill_rxs = vec![];
+    for _ in 0..num_workers {
+        killpill_rxs.push(rx.resubscribe());
+    }
+
+    if rx.try_recv().is_ok() {
+        tracing::info!("Received killpill, exiting");
+        return Ok(());
+    }
    let instance_name = gethostname()
        .to_str()
        .map(|x| {
@@ -673,17 +680,15 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
            .expect("could not create initial worker dir");
    }

-    let sync_barrier = Arc::new(RwLock::new(None));
    for i in 1..(num_workers + 1) {
        let db1 = db.clone();
        let instance_name = instance_name.clone();
        let worker_name = format!("wk-{}-{}-{}", *WORKER_GROUP, &instance_name, rd_string(5));
        let ip = ip.clone();
-        let rx = rx.resubscribe();
+        let rx = killpill_rxs.pop().unwrap();
        let tx = tx.clone();
        let base_internal_url = base_internal_url.clone();
        let rsmq2 = rsmq.clone();
-        let sync_barrier = sync_barrier.clone();

        handles.push(tokio::spawn(async move {
            tracing::info!(worker = %worker_name, "starting worker");
@@ -699,7 +704,6 @@ pub async fn run_workers<R: rsmq_async::RsmqConnection + Send + Sync + Clone + '
                tx,
                &base_internal_url,
                rsmq2,
-                sync_barrier,
                agent_mode,
            );

@@ -1,4 +1,3 @@
-use std::sync::Arc;
 use windmill_api_client::types::{NewScript, NewScriptLanguage};
 use std::str::FromStr;

@@ -11,7 +10,6 @@ use serde::Deserialize;
 use serde_json::json;
 use sqlx::types::Json;
 use sqlx::{postgres::PgListener, types::Uuid, Pool, Postgres};
-use tokio::sync::RwLock;

 #[cfg(feature = "enterprise")]
 use tokio::time::{timeout, Duration};
@@ -993,7 +991,6 @@ fn spawn_test_worker(
            tx2,
            &base_internal_url,
            None,
-            Arc::new(RwLock::new(None)),
            false,
        )
        .await
@@ -71,7 +71,7 @@ use tokio::{
    fs::DirBuilder,
    sync::{
        mpsc::{self, Sender},
-        Barrier, RwLock,
+        RwLock,
    },
    task::JoinHandle,
    time::Instant,
@@ -614,7 +614,6 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
    killpill_tx: tokio::sync::broadcast::Sender<()>,
    base_internal_url: &str,
    rsmq: Option<R>,
-    _sync_barrier: Arc<RwLock<Option<Barrier>>>,
    agent_mode: bool,
 ) {
    #[cfg(not(feature = "enterprise"))]