pharos_native/runtime/
job.rs

1use std::sync::Arc;
2#[cfg(not(target_os = "wasi"))]
3use std::sync::Mutex;
4#[cfg(not(target_os = "wasi"))]
5use std::sync::atomic::AtomicU8;
6use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
7#[cfg(not(target_os = "wasi"))]
8use std::thread::{self, JoinHandle};
9#[cfg(not(target_os = "wasi"))]
10use std::time::Duration;
11
12#[cfg(not(target_os = "wasi"))]
13const WORKER_STOPPED: u8 = 0;
14#[cfg(not(target_os = "wasi"))]
15const WORKER_STARTING: u8 = 1;
16#[cfg(not(target_os = "wasi"))]
17const WORKER_RUNNING: u8 = 2;
18#[cfg(not(target_os = "wasi"))]
19const WORKER_STOPPING: u8 = 3;
20
21#[derive(Clone)]
22/// Shared state for one background job worker.
23///
24/// # Fields
25///
26/// - This type stores module-owned state for [`crate::runtime::job::JobWorkerState`].
27///
28/// # Examples
29///
30/// ```rust,ignore
31/// let _ = std::mem::size_of::<pharos_native::runtime::job::JobWorkerState>();
32/// ```
33///
34/// # Related
35///
36/// - [`crate::runtime::job`] for constructors, helpers, and neighboring runtime types.
37pub struct JobWorkerState {
38    running: Arc<AtomicBool>,
39    shutdown: Arc<AtomicBool>,
40    poll_interval_ms: u64,
41    poll_count: Arc<AtomicUsize>,
42    poll_handler: Arc<dyn Fn() + Send + Sync + 'static>,
43    #[cfg(not(target_os = "wasi"))]
44    lifecycle: Arc<AtomicU8>,
45    #[cfg(not(target_os = "wasi"))]
46    thread: Arc<Mutex<Option<JoinHandle<()>>>>,
47}
48
49impl Default for JobWorkerState {
50    /// Returns the default default.
51    fn default() -> Self {
52        Self {
53            running: Arc::new(AtomicBool::new(false)),
54            shutdown: Arc::new(AtomicBool::new(false)),
55            poll_interval_ms: 5_000,
56            poll_count: Arc::new(AtomicUsize::new(0)),
57            poll_handler: Arc::new(|| {}),
58            #[cfg(not(target_os = "wasi"))]
59            lifecycle: Arc::new(AtomicU8::new(WORKER_STOPPED)),
60            #[cfg(not(target_os = "wasi"))]
61            thread: Arc::new(Mutex::new(None)),
62        }
63    }
64}
65
66impl JobWorkerState {
67    /// Configures the interval between completed poll cycles.
68    ///
69    /// # Parameters
70    ///
71    /// - `self`: Existing worker state being reconfigured.
72    /// - `poll_interval_ms`: Milliseconds to wait between completed poll cycles.
73    ///
74    /// # Returns
75    ///
76    /// Returns the updated worker state.
77    ///
78    /// # Examples
79    ///
80    /// ```rust,ignore
81    /// let _result = pharos_native::runtime::job::JobWorkerState::default().with_poll_interval_ms(1000);
82    /// ```
83    ///
84    /// # Related
85    ///
86    /// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
87    pub fn with_poll_interval_ms(mut self, poll_interval_ms: u64) -> Self {
88        self.poll_interval_ms = poll_interval_ms;
89        self
90    }
91
92    /// Connects each poll cycle to the durable runtime job dispatcher.
93    ///
94    /// # Parameters
95    ///
96    /// - `self`: Existing worker state being reconfigured.
97    /// - `poll_handler`: Handler executed once per poll cycle.
98    ///
99    /// # Returns
100    ///
101    /// Returns the updated worker state.
102    ///
103    /// # Examples
104    ///
105    /// ```rust,ignore
106    /// let _result = pharos_native::runtime::job::JobWorkerState::default().with_poll_handler(std::sync::Arc::new(|| {}));
107    /// ```
108    ///
109    /// # Related
110    ///
111    /// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
112    pub fn with_poll_handler(
113        mut self,
114        poll_handler: Arc<dyn Fn() + Send + Sync + 'static>,
115    ) -> Self {
116        self.poll_handler = poll_handler;
117        self
118    }
119
120    /// Requests a bounded shutdown at the next poll or sleep boundary.
121    ///
122    /// # Returns
123    ///
124    /// Returns after the shutdown flag is raised for the worker.
125    ///
126    /// # Examples
127    ///
128    /// ```rust,ignore
129    /// pharos_native::runtime::job::JobWorkerState::default().request_shutdown();
130    /// ```
131    ///
132    /// # Related
133    ///
134    /// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
135    pub fn request_shutdown(&self) {
136        self.shutdown.store(true, Ordering::SeqCst);
137        #[cfg(not(target_os = "wasi"))]
138        {
139            let _ = self.lifecycle.compare_exchange(
140                WORKER_RUNNING,
141                WORKER_STOPPING,
142                Ordering::SeqCst,
143                Ordering::SeqCst,
144            );
145            let _ = self.lifecycle.compare_exchange(
146                WORKER_STARTING,
147                WORKER_STOPPING,
148                Ordering::SeqCst,
149                Ordering::SeqCst,
150            );
151        }
152    }
153
154    /// Returns the number of poll cycles completed by this state.
155    ///
156    /// # Returns
157    ///
158    /// Returns the total completed poll count observed by this worker state.
159    ///
160    /// # Examples
161    ///
162    /// ```rust,ignore
163    /// let _result = pharos_native::runtime::job::JobWorkerState::default().poll_count();
164    /// ```
165    ///
166    /// # Related
167    ///
168    /// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
169    pub fn poll_count(&self) -> usize {
170        self.poll_count.load(Ordering::SeqCst)
171    }
172}
173
174#[cfg(not(target_os = "wasi"))]
175#[derive(Debug, Clone, Copy, Eq, PartialEq)]
176/// Error returned when a managed worker lifecycle operation cannot complete.
177///
178/// # Variants
179///
180/// - `AlreadyRunning`: Supported `JobWorkerLifecycleError` variant.
181/// - `ThreadSpawnFailed`: Supported `JobWorkerLifecycleError` variant.
182/// - `ThreadPanicked`: Supported `JobWorkerLifecycleError` variant.
183/// - `LifecyclePoisoned`: Supported `JobWorkerLifecycleError` variant.
184///
185/// # Examples
186///
187/// ```rust,ignore
188/// let _ = pharos_native::runtime::job::JobWorkerLifecycleError::AlreadyRunning;
189/// ```
190///
191/// # Related
192///
193/// - [`crate::runtime::job`] for constructors, helpers, and neighboring runtime types.
194pub enum JobWorkerLifecycleError {
195    /// A worker is already starting or running.
196    AlreadyRunning,
197    /// The worker thread could not be created.
198    ThreadSpawnFailed,
199    /// The worker thread panicked before it could be joined cleanly.
200    ThreadPanicked,
201    /// A prior lifecycle operation poisoned the lifecycle mutex.
202    LifecyclePoisoned,
203}
204
205/// Returns whether the worker has entered its running state.
206///
207/// # Parameters
208///
209/// - `state`: Input consumed by [`crate::runtime::job::job_worker_is_running`].
210///
211/// # Returns
212///
213/// Returns `true` when the condition checked by [`crate::runtime::job::job_worker_is_running`] holds and `false` otherwise.
214///
215/// # Examples
216///
217/// ```rust,ignore
218/// let _result = pharos_native::runtime::job::job_worker_is_running(todo!());
219/// ```
220///
221/// # Related
222///
223/// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
224pub fn job_worker_is_running(state: &JobWorkerState) -> bool {
225    state.running.load(Ordering::SeqCst)
226}
227
228#[cfg(not(target_os = "wasi"))]
229/// Makes an owned Rust copy of an optional C-style string value.
230///
231/// # Parameters
232///
233/// - `value`: Input consumed by [`crate::runtime::job::job_worker_strdup_local`].
234///
235/// # Returns
236///
237/// Returns [`Some`] when [`crate::runtime::job::job_worker_strdup_local`] can produce a value and [`None`] otherwise.
238///
239/// # Examples
240///
241/// ```rust,ignore
242/// let _result = pharos_native::runtime::job::job_worker_strdup_local("value");
243/// ```
244///
245/// # Related
246///
247/// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
248pub fn job_worker_strdup_local(value: Option<&str>) -> Option<String> {
249    value.map(ToOwned::to_owned)
250}
251
252#[cfg(not(target_os = "wasi"))]
253/// Runs one bounded durable job-dispatch poll operation.
254///
255/// # Parameters
256///
257/// - `state`: Input consumed by [`crate::runtime::job::job_worker_poll_once`].
258///
259/// # Returns
260///
261/// Returns after [`crate::runtime::job::job_worker_poll_once`] applies its in-place effect.
262///
263/// # Examples
264///
265/// ```rust,ignore
266/// let _result = pharos_native::runtime::job::job_worker_poll_once(todo!());
267/// ```
268///
269/// # Related
270///
271/// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
272pub fn job_worker_poll_once(state: &JobWorkerState) {
273    (state.poll_handler)();
274    state.poll_count.fetch_add(1, Ordering::SeqCst);
275}
276
277#[cfg(not(target_os = "wasi"))]
278/// Sleeps in bounded slices and reports whether shutdown was not requested.
279///
280/// # Parameters
281///
282/// - `state`: Input consumed by [`crate::runtime::job::job_worker_sleep_or_shutdown`].
283/// - `milliseconds`: Input consumed by [`crate::runtime::job::job_worker_sleep_or_shutdown`].
284///
285/// # Returns
286///
287/// Returns `true` when the condition checked by [`crate::runtime::job::job_worker_sleep_or_shutdown`] holds and `false` otherwise.
288///
289/// # Examples
290///
291/// ```rust,ignore
292/// let _result = pharos_native::runtime::job::job_worker_sleep_or_shutdown(todo!(), todo!());
293/// ```
294///
295/// # Related
296///
297/// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
298pub fn job_worker_sleep_or_shutdown(state: &JobWorkerState, milliseconds: u64) -> bool {
299    if state.shutdown.load(Ordering::SeqCst) {
300        return false;
301    }
302    if milliseconds == 0 {
303        thread::yield_now();
304        return !state.shutdown.load(Ordering::SeqCst);
305    }
306
307    let mut remaining = Duration::from_millis(milliseconds);
308    while !remaining.is_zero() {
309        if state.shutdown.load(Ordering::SeqCst) {
310            return false;
311        }
312        let slice = remaining.min(Duration::from_millis(100));
313        let started = std::time::Instant::now();
314        thread::sleep(slice);
315        let elapsed = started.elapsed();
316        remaining = remaining.saturating_sub(elapsed);
317    }
318    !state.shutdown.load(Ordering::SeqCst)
319}
320
321#[cfg(not(target_os = "wasi"))]
322/// Runs the worker until shutdown or the optional completed-poll limit.
323///
324/// # Parameters
325///
326/// - `state`: Input consumed by [`crate::runtime::job::job_worker_thread_main`].
327/// - `max_polls`: Input consumed by [`crate::runtime::job::job_worker_thread_main`].
328///
329/// # Returns
330///
331/// Returns after [`crate::runtime::job::job_worker_thread_main`] applies its in-place effect.
332///
333/// # Examples
334///
335/// ```rust,ignore
336/// let _result = pharos_native::runtime::job::job_worker_thread_main(todo!(), todo!());
337/// ```
338///
339/// # Related
340///
341/// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
342pub fn job_worker_thread_main(state: JobWorkerState, max_polls: Option<usize>) {
343    if state.shutdown.load(Ordering::SeqCst) {
344        state.running.store(false, Ordering::SeqCst);
345        state.lifecycle.store(WORKER_STOPPED, Ordering::SeqCst);
346        return;
347    }
348
349    let entered_running = state
350        .lifecycle
351        .compare_exchange(
352            WORKER_STARTING,
353            WORKER_RUNNING,
354            Ordering::SeqCst,
355            Ordering::SeqCst,
356        )
357        .is_ok()
358        || state
359            .lifecycle
360            .compare_exchange(
361                WORKER_STOPPED,
362                WORKER_RUNNING,
363                Ordering::SeqCst,
364                Ordering::SeqCst,
365            )
366            .is_ok();
367    if !entered_running || state.shutdown.load(Ordering::SeqCst) {
368        state.running.store(false, Ordering::SeqCst);
369        state.lifecycle.store(WORKER_STOPPED, Ordering::SeqCst);
370        return;
371    }
372
373    state.running.store(true, Ordering::SeqCst);
374    while !state.shutdown.load(Ordering::SeqCst) {
375        if max_polls.is_some_and(|limit| state.poll_count() >= limit) {
376            break;
377        }
378        job_worker_poll_once(&state);
379        if max_polls.is_some_and(|limit| state.poll_count() >= limit) {
380            break;
381        }
382        if !job_worker_sleep_or_shutdown(&state, state.poll_interval_ms) {
383            break;
384        }
385    }
386
387    state.running.store(false, Ordering::SeqCst);
388    state.lifecycle.store(WORKER_STOPPED, Ordering::SeqCst);
389}
390
391#[cfg(not(target_os = "wasi"))]
392/// Starts one managed worker thread for the supplied state.
393///
394/// # Parameters
395///
396/// - `state`: Input consumed by [`crate::runtime::job::job_worker_start`].
397/// - `max_polls`: Input consumed by [`crate::runtime::job::job_worker_start`].
398/// - `) -> Result<(`: Input consumed by [`crate::runtime::job::job_worker_start`].
399///
400/// # Returns
401///
402/// Returns the successful value produced by [`crate::runtime::job::job_worker_start`] or an error when the operation fails.
403///
404/// # Errors
405///
406/// Returns an error value when [`crate::runtime::job::job_worker_start`] cannot complete successfully.
407///
408/// # Examples
409///
410/// ```rust,ignore
411/// let _result = pharos_native::runtime::job::job_worker_start(todo!(), todo!(), todo!());
412/// ```
413///
414/// # Related
415///
416/// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
417pub fn job_worker_start(
418    state: &JobWorkerState,
419    max_polls: Option<usize>,
420) -> Result<(), JobWorkerLifecycleError> {
421    let mut thread_slot = state
422        .thread
423        .lock()
424        .map_err(|_| JobWorkerLifecycleError::LifecyclePoisoned)?;
425    if thread_slot.is_some()
426        || state
427            .lifecycle
428            .compare_exchange(
429                WORKER_STOPPED,
430                WORKER_STARTING,
431                Ordering::SeqCst,
432                Ordering::SeqCst,
433            )
434            .is_err()
435    {
436        return Err(JobWorkerLifecycleError::AlreadyRunning);
437    }
438
439    state.shutdown.store(false, Ordering::SeqCst);
440    state.running.store(true, Ordering::SeqCst);
441    let worker_state = state.clone();
442    let spawned = thread::Builder::new()
443        .name("pharos-job-worker".to_owned())
444        .spawn(move || job_worker_thread_main(worker_state, max_polls));
445    let handle = match spawned {
446        Ok(handle) => handle,
447        Err(_) => {
448            state.running.store(false, Ordering::SeqCst);
449            state.lifecycle.store(WORKER_STOPPED, Ordering::SeqCst);
450            return Err(JobWorkerLifecycleError::ThreadSpawnFailed);
451        }
452    };
453
454    *thread_slot = Some(handle);
455    Ok(())
456}
457
458#[cfg(not(target_os = "wasi"))]
459/// Signals a managed worker, joins it, and releases its owned thread handle.
460///
461/// # Parameters
462///
463/// - `state`: Input consumed by [`crate::runtime::job::job_worker_stop`].
464///
465/// # Returns
466///
467/// Returns the successful value produced by [`crate::runtime::job::job_worker_stop`] or an error when the operation fails.
468///
469/// # Errors
470///
471/// Returns an error value when [`crate::runtime::job::job_worker_stop`] cannot complete successfully.
472///
473/// # Examples
474///
475/// ```rust,ignore
476/// let _result = pharos_native::runtime::job::job_worker_stop(todo!());
477/// ```
478///
479/// # Related
480///
481/// - [`crate::runtime::job`] for adjacent helpers that share the same runtime surface.
482pub fn job_worker_stop(state: &JobWorkerState) -> Result<(), JobWorkerLifecycleError> {
483    let handle = {
484        let mut thread_slot = state
485            .thread
486            .lock()
487            .map_err(|_| JobWorkerLifecycleError::LifecyclePoisoned)?;
488        if thread_slot.is_none() && !job_worker_is_running(state) {
489            state.shutdown.store(false, Ordering::SeqCst);
490            state.lifecycle.store(WORKER_STOPPED, Ordering::SeqCst);
491            return Ok(());
492        }
493        state.shutdown.store(true, Ordering::SeqCst);
494        state.lifecycle.store(WORKER_STOPPING, Ordering::SeqCst);
495        thread_slot.take()
496    };
497
498    let join_result = handle.map(|worker| worker.join());
499    state.running.store(false, Ordering::SeqCst);
500    state.shutdown.store(false, Ordering::SeqCst);
501    state.lifecycle.store(WORKER_STOPPED, Ordering::SeqCst);
502    match join_result {
503        Some(Ok(())) | None => Ok(()),
504        Some(Err(_)) => Err(JobWorkerLifecycleError::ThreadPanicked),
505    }
506}
pharos_native/runtime/job.rs

pharos_native/runtime/
job.rs
