zng_task/process/
worker.rs

1//! Async worker process tasks.
2//!
3//! This module defines a worker process that can run tasks in a separate process instance.
4//!
5//! Each worker process can run multiple tasks in parallel, the worker type is [`Worker`]. Note that this module does not offer a fork
6//! implementation, the worker processes begin from the start state. The primary use of process tasks is to make otherwise fatal tasks
7//! recoverable, if the task calls unsafe code or code that can potentially terminate the entire process it should run using a [`Worker`].
8//! If you only want to recover from panics in safe code consider using [`task::run_catch`] or [`task::wait_catch`] instead.
9//!
10//! You can send [IPC channel] endpoints in the task request messages, this can be useful for implementing progress reporting,
11//! you can also send [`IpcBytes`] to efficiently share large byte blobs with the worker process.
12//!
13//! [`task::run_catch`]: crate::run_catch
14//! [`task::wait_catch`]: crate::wait_catch
15//! [IPC channel]: crate::channel::ipc_unbounded
16//! [`IpcBytes`]: crate::channel::IpcBytes
17//!
18//! # Examples
19//!
20//! The example below demonstrates a worker-process setup that uses the same executable as the app-process.
21//!
22//! ```
23//! # fn main() { }
24//! # mod zng { pub mod env { pub use zng_env::*; } pub mod task { pub use zng_task::*; } }
25//! # fn demo() {
26//! fn main() {
27//!     zng::env::init!();
28//!     // normal app init..
29//!     # zng::task::doc_test(false, on_click());
30//! }
31//! # }
32//!
33//! mod task1 {
34//! # use super::zng;
35//!     use zng::{task::process::worker, env};
36//!
37//!     const NAME: &str = "zng::example::task1";
38//!
39//!     env::on_process_start!(|args| {
40//!         // give tracing handlers a chance to observe the worker-process
41//!         if args.yield_count == 0 { return args.yield_once(); }
42//          // run the worker server
43//!         worker::run_worker(NAME, work);
44//!     });
45//!     async fn work(args: worker::RequestArgs<Request>) -> Response {
46//!         let rsp = format!("received 'task1' request `{:?}` in worker-process #{}", &args.request.data, std::process::id());
47//!         Response { data: rsp }
48//!     }
49//!     
50//!     #[derive(Debug, serde::Serialize, serde::Deserialize)]
51//!     pub struct Request { pub data: String }
52//!
53//!     #[derive(Debug, serde::Serialize, serde::Deserialize)]
54//!     pub struct Response { pub data: String }
55//!
56//!     // called in app-process
57//!     pub async fn start() -> worker::Worker<Request, Response> {
58//!         worker::Worker::start(NAME).await.expect("cannot spawn 'task1'")
59//!     }
60//! }
61//!
62//! // This runs in the app-process, it starts a worker process and requests a task run.
63//! async fn on_click() {
64//!     println!("app-process #{} starting a worker", std::process::id());
65//!     let mut worker = task1::start().await;
66//!     // request a task run and await it.
67//!     match worker.run(task1::Request { data: "request".to_owned() }).await {
68//!         Ok(task1::Response { data }) => println!("ok. {data}"),
69//!         Err(e) => eprintln!("error: {e}"),
70//!     }
71//!     // multiple tasks can be requested in parallel, use `task::all!` to await ..
72//!
73//!     // the worker process can be gracefully shutdown, awaits all pending tasks.
74//!     let _ = worker.shutdown().await;
75//! }
76//! ```
77//!
78//! Note that you can setup multiple workers the same executable, as long as the `on_process_start!` call happens
79//! on different modules.
80//!
81//! # Connect Timeout
82//!
83//! If the worker process takes longer than 10 seconds to connect the tasks fails. This is more then enough in most cases, but
84//! it can be too little in some test runner machines. You can set the `"ZNG_TASK_WORKER_TIMEOUT"` environment variable to a custom
85//! timeout in seconds. The minimum value is 1 second, set to 0 or empty use the default timeout.
86
87use core::fmt;
88use std::{marker::PhantomData, path::PathBuf, pin::Pin, sync::Arc};
89
90use parking_lot::Mutex;
91use zng_clone_move::{async_clmv, clmv};
92use zng_txt::Txt;
93use zng_unique_id::IdMap;
94use zng_unit::TimeUnits as _;
95
96use crate::{
97    TaskPanicError,
98    channel::{self, ChannelError, IpcReceiver, IpcSender, IpcValue, NamedIpcSender},
99};
100
101const WORKER_VERSION: &str = "ZNG_TASK_IPC_WORKER_VERSION";
102const WORKER_SERVER: &str = "ZNG_TASK_IPC_WORKER_SERVER";
103const WORKER_NAME: &str = "ZNG_TASK_IPC_WORKER_NAME";
104
105const WORKER_TIMEOUT: &str = "ZNG_TASK_WORKER_TIMEOUT";
106
107/// The *App Process* and *Worker Process* must be build using the same exact version and this is
108/// validated during run-time, causing a panic if the versions don't match.
109pub const VERSION: &str = env!("CARGO_PKG_VERSION");
110
111/// Represents a running worker process.
112pub struct Worker<I: IpcValue, O: IpcValue> {
113    running: Option<(std::thread::JoinHandle<()>, std::process::Child)>,
114
115    sender: IpcSender<(RequestId, Request<I>)>,
116    requests: Arc<Mutex<IdMap<RequestId, channel::Sender<O>>>>,
117
118    _p: PhantomData<fn(I) -> O>,
119
120    crash: Option<WorkerCrashError>,
121}
122impl<I: IpcValue, O: IpcValue> Worker<I, O> {
123    /// Start a worker process implemented in the current executable.
124    ///
125    /// Note that the current process must call [`run_worker`] at startup to actually work.
126    /// You can use [`zng_env::on_process_start!`] to inject startup code.
127    pub async fn start(worker_name: impl Into<Txt>) -> std::io::Result<Self> {
128        Self::start_impl(worker_name.into(), std::env::current_exe()?, &[], &[]).await
129    }
130
131    /// Start a worker process implemented in the current executable with custom env vars and args.
132    pub async fn start_with(worker_name: impl Into<Txt>, env_vars: &[(&str, &str)], args: &[&str]) -> std::io::Result<Self> {
133        Self::start_impl(worker_name.into(), std::env::current_exe()?, env_vars, args).await
134    }
135
136    /// Start a worker process implemented in another executable with custom env vars and args.
137    pub async fn start_other(
138        worker_name: impl Into<Txt>,
139        worker_exe: impl Into<PathBuf>,
140        env_vars: &[(&str, &str)],
141        args: &[&str],
142    ) -> std::io::Result<Self> {
143        Self::start_impl(worker_name.into(), worker_exe.into(), env_vars, args).await
144    }
145
146    async fn start_impl(worker_name: Txt, exe: PathBuf, env_vars: &[(&str, &str)], args: &[&str]) -> std::io::Result<Self> {
147        let chan_sender = NamedIpcSender::<WorkerInit<I, O>>::new()?;
148
149        let mut worker = std::process::Command::new(dunce::canonicalize(exe)?);
150        for (key, value) in env_vars {
151            worker.env(key, value);
152        }
153        for arg in args {
154            worker.arg(arg);
155        }
156        worker
157            .env(WORKER_VERSION, crate::process::worker::VERSION)
158            .env(WORKER_SERVER, chan_sender.name())
159            .env(WORKER_NAME, worker_name)
160            .env("RUST_BACKTRACE", "full");
161        let mut worker = blocking::unblock(move || worker.spawn()).await?;
162
163        let timeout = match std::env::var(WORKER_TIMEOUT) {
164            Ok(t) if !t.is_empty() => match t.parse::<u64>() {
165                Ok(t) => t.max(1),
166                Err(e) => {
167                    tracing::error!("invalid {WORKER_TIMEOUT:?} value, {e}");
168                    10
169                }
170            },
171            _ => 10,
172        };
173
174        let (request_sender, mut response_receiver) = match Self::connect_worker(chan_sender, timeout).await {
175            Ok(r) => r,
176            Err(ce) => {
177                let cleanup = blocking::unblock(move || {
178                    worker.kill()?;
179                    worker.wait()
180                });
181                match cleanup.await {
182                    Ok(status) => {
183                        let code = status.code().unwrap_or(0);
184                        return Err(std::io::Error::new(
185                            std::io::ErrorKind::TimedOut,
186                            format!("worker process did not connect in {timeout}s\nworker exit code: {code}\nchannel error: {ce}"),
187                        ));
188                    }
189                    Err(e) => {
190                        return Err(std::io::Error::new(
191                            std::io::ErrorKind::TimedOut,
192                            format!("worker process did not connect in {timeout}s\ncannot kill worker process, {e}\nchannel error: {ce}"),
193                        ));
194                    }
195                }
196            }
197        };
198
199        let requests = Arc::new(Mutex::new(IdMap::<RequestId, channel::Sender<O>>::new()));
200        let receiver = std::thread::Builder::new()
201            .name("task-ipc-recv".into())
202            .stack_size(256 * 1024)
203            .spawn(clmv!(requests, || {
204                loop {
205                    match response_receiver.recv_blocking() {
206                        Ok((id, r)) => match requests.lock().remove(&id) {
207                            Some(s) => match r {
208                                Response::Out(r) => {
209                                    let _ = s.send_blocking(r);
210                                }
211                            },
212                            None => tracing::error!("worker responded to unknown request #{}", id.sequential()),
213                        },
214                        Err(e) => match e {
215                            ChannelError::Disconnected { .. } => {
216                                requests.lock().clear();
217                                break;
218                            }
219                            e => {
220                                tracing::error!("worker response error, will shutdown, {e}");
221                                break;
222                            }
223                        },
224                    }
225                }
226            }))
227            .expect("failed to spawn thread");
228
229        Ok(Self {
230            running: Some((receiver, worker)),
231            sender: request_sender,
232            _p: PhantomData,
233            crash: None,
234            requests,
235        })
236    }
237    async fn connect_worker(
238        chan_sender: NamedIpcSender<WorkerInit<I, O>>,
239        timeout: u64,
240    ) -> Result<(IpcSender<(RequestId, Request<I>)>, IpcReceiver<(RequestId, Response<O>)>), ChannelError> {
241        let mut chan_sender = chan_sender.connect_deadline(timeout.secs()).await?;
242
243        let (request_sender, request_receiver) =
244            channel::ipc_unbounded::<(RequestId, Request<I>)>().map_err(ChannelError::disconnected_by)?;
245        let (response_sender, response_receiver) =
246            channel::ipc_unbounded::<(RequestId, Response<O>)>().map_err(ChannelError::disconnected_by)?;
247
248        chan_sender.send_blocking((request_receiver, response_sender))?;
249
250        Ok((request_sender, response_receiver))
251    }
252
253    /// Awaits current tasks and kills the worker process.
254    pub async fn shutdown(mut self) -> std::io::Result<()> {
255        if let Some((receiver, mut process)) = self.running.take() {
256            while !self.requests.lock().is_empty() {
257                crate::deadline(100.ms()).await;
258            }
259            let r = blocking::unblock(move || process.kill()).await;
260
261            match crate::with_deadline(blocking::unblock(move || receiver.join()), 1.secs()).await {
262                Ok(r) => {
263                    if let Err(p) = r {
264                        tracing::error!(
265                            "worker receiver thread exited panicked, {}",
266                            TaskPanicError::new(p).panic_str().unwrap_or("")
267                        );
268                    }
269                }
270                Err(_) => {
271                    // timeout
272                    if r.is_ok() {
273                        // after awaiting kill receiver thread should join fast because disconnect breaks loop
274                        panic!("worker receiver thread did not exit after worker process did");
275                    }
276                }
277            }
278            r
279        } else {
280            Ok(())
281        }
282    }
283
284    /// Run a task in a free worker process thread.
285    pub fn run(&mut self, input: I) -> impl Future<Output = Result<O, RunError>> + Send + 'static {
286        self.run_request(Request::Run(input))
287    }
288
289    fn run_request(&mut self, request: Request<I>) -> Pin<Box<dyn Future<Output = Result<O, RunError>> + Send + 'static>> {
290        if self.crash_error().is_some() {
291            return Box::pin(std::future::ready(Err(RunError::Disconnected)));
292        }
293
294        let id = RequestId::new_unique();
295        let (sx, rx) = channel::bounded(1);
296
297        let requests = self.requests.clone();
298        requests.lock().insert(id, sx);
299        let mut sender = self.sender.clone();
300        let send_r = blocking::unblock(move || sender.send_blocking((id, request)));
301
302        Box::pin(async move {
303            if let Err(e) = send_r.await {
304                requests.lock().remove(&id);
305                return Err(RunError::Other(Arc::new(e)));
306            }
307
308            match rx.recv().await {
309                Ok(r) => Ok(r),
310                Err(e) => match e {
311                    ChannelError::Disconnected { .. } => {
312                        requests.lock().remove(&id);
313                        Err(RunError::Disconnected)
314                    }
315                    _ => unreachable!(),
316                },
317            }
318        })
319    }
320
321    /// Reference the crash error.
322    ///
323    /// The worker cannot be used if this is set, run requests will immediately disconnect.
324    pub fn crash_error(&mut self) -> Option<&WorkerCrashError> {
325        if let Some((t, _)) = &self.running
326            && t.is_finished()
327        {
328            let (t, mut p) = self.running.take().unwrap();
329
330            if let Err(e) = t.join() {
331                tracing::error!(
332                    "panic in worker receiver thread, {}",
333                    TaskPanicError::new(e).panic_str().unwrap_or("")
334                );
335            }
336
337            if let Err(e) = p.kill() {
338                tracing::error!("error killing worker process after receiver exit, {e}");
339            }
340
341            match p.wait() {
342                Ok(o) => {
343                    self.crash = Some(WorkerCrashError { status: o });
344                }
345                Err(e) => tracing::error!("error reading crashed worker output, {e}"),
346            }
347        }
348
349        self.crash.as_ref()
350    }
351}
352impl<I: IpcValue, O: IpcValue> Drop for Worker<I, O> {
353    fn drop(&mut self) {
354        if let Some((receiver, mut process)) = self.running.take() {
355            if !receiver.is_finished() {
356                tracing::error!("dropped worker without shutdown");
357            }
358            if let Err(e) = process.kill() {
359                tracing::error!("failed to kill worker process on drop, {e}");
360            }
361        }
362    }
363}
364
365/// If the process was started by a [`Worker`] runs the worker loop and never returns. If
366/// not started as worker does nothing.
367///
368/// The `handler` is called for each work request.
369pub fn run_worker<I, O, F>(worker_name: impl Into<Txt>, handler: impl Fn(RequestArgs<I>) -> F + Send + Sync + 'static)
370where
371    I: IpcValue,
372    O: IpcValue,
373    F: Future<Output = O> + Send + Sync + 'static,
374{
375    let name = worker_name.into();
376    if let Some(server_name) = run_worker_server(&name) {
377        zng_env::init_process_name(zng_txt::formatx!("worker-process ({name}, {})", std::process::id()));
378
379        let mut chan_recv = IpcReceiver::<WorkerInit<I, O>>::connect(server_name)
380            .unwrap_or_else(|e| panic!("failed to connect to '{name}' init channel, {e}"));
381
382        let (mut request_receiver, response_sender) = chan_recv
383            .recv_blocking()
384            .unwrap_or_else(|e| panic!("failed to connect initial channels, {e}"));
385
386        let handler = Arc::new(handler);
387
388        loop {
389            match request_receiver.recv_blocking() {
390                Ok((id, input)) => match input {
391                    Request::Run(r) => crate::spawn(async_clmv!(handler, mut response_sender, {
392                        let output = handler(RequestArgs { request: r }).await;
393                        let _ = response_sender.send_blocking((id, Response::Out(output)));
394                    })),
395                },
396                Err(e) => match e {
397                    ChannelError::Disconnected { .. } => break,
398                    ChannelError::Timeout => unreachable!(),
399                },
400            }
401        }
402
403        zng_env::exit(0);
404    }
405}
406fn run_worker_server(worker_name: &str) -> Option<String> {
407    if let Ok(w_name) = std::env::var(WORKER_NAME)
408        && let Ok(version) = std::env::var(WORKER_VERSION)
409        && let Ok(server_name) = std::env::var(WORKER_SERVER)
410    {
411        if w_name != worker_name {
412            return None;
413        }
414        if version != VERSION {
415            eprintln!("worker '{worker_name}' API version is not equal, app-process: {version}, worker-process: {VERSION}");
416            zng_env::exit(i32::from_le_bytes(*b"vapi"));
417        }
418
419        Some(server_name)
420    } else {
421        None
422    }
423}
424
425/// Arguments for [`run_worker`].
426#[non_exhaustive]
427pub struct RequestArgs<I: IpcValue> {
428    /// The task request data.
429    pub request: I,
430}
431
432/// Worker run error.
433#[derive(Debug, Clone)]
434#[non_exhaustive]
435pub enum RunError {
436    /// Lost connection with the worker process.
437    ///
438    /// See [`Worker::crash_error`] for the error.
439    Disconnected,
440    /// Other error.
441    Other(Arc<dyn std::error::Error + Send + Sync>),
442}
443impl fmt::Display for RunError {
444    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
445        match self {
446            RunError::Disconnected => write!(f, "worker process disconnected"),
447            RunError::Other(e) => write!(f, "run error, {e}"),
448        }
449    }
450}
451impl std::error::Error for RunError {}
452
453/// Info about a worker process crash.
454#[derive(Debug, Clone)]
455#[non_exhaustive]
456pub struct WorkerCrashError {
457    /// Worker process exit code.
458    pub status: std::process::ExitStatus,
459}
460impl fmt::Display for WorkerCrashError {
461    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
462        write!(f, "{:?}", self.status)
463    }
464}
465impl std::error::Error for WorkerCrashError {}
466
467#[derive(serde::Serialize, serde::Deserialize)]
468enum Request<I> {
469    Run(I),
470}
471
472#[derive(serde::Serialize, serde::Deserialize)]
473enum Response<O> {
474    Out(O),
475}
476
477type WorkerInit<I, O> = (
478    channel::IpcReceiver<(RequestId, Request<I>)>,
479    channel::IpcSender<(RequestId, Response<O>)>,
480);
481
482zng_unique_id::unique_id_64! {
483    #[derive(serde::Serialize, serde::Deserialize)]
484    struct RequestId;
485}
zng_task/process/worker.rs

zng_task/process/
worker.rs
