sound_to_patch/render/
session.rs

1//! The render session: the one worker pool that calibration, refine, the
2//! bench and the dataset render through. None of them owns a pool.
3//!
4//! Each worker holds one renderer, and so one plug-in instance in its own
5//! host process. A worker starts its renderer at its first job, and starts
6//! a new one after [`RESTART_AFTER`] renders, so state a plug-in builds up
7//! over many loads cannot reach later renders. A failed render is retried
8//! once on a new renderer. A second failure gives a [`RenderFailure`],
9//! which a consumer scores [`FAILED_SCORE`]; the session goes on with the
10//! next job.
11//!
12//! Every render starts with the flush of the recipe. That is the
13//! renderer's duty, not the session's: every [`Renderer::render`] loads
14//! the job's state and plays the pre-roll before the clip.
15//!
16//! A panic in a renderer counts as a failed try, so it costs one job a
17//! retry and never the jobs of the other workers. A
18//! [`RenderError::WrongSynth`] is the caller's mistake, not a fault of
19//! the plug-in: it fails the job at once and keeps the renderer.
20//!
21//! # Repeat renders
22//!
23//! A fresh Vital instance renders bit-identically to any other fresh
24//! instance, and each later render on one instance differs from the ones
25//! before. So the n-th renders of two workers are equal. A consumer that
26//! needs repeat-render noise draws its renders in sequence on one
27//! instance. One that needs a repeatable render renders on a fresh
28//! instance.
29//!
30//! # Costs
31//!
32//! - Each worker holds one plug-in in its own host process. A Vital host
33//!   takes about 270 to 325 MB of resident memory, and the session sets no
34//!   cap on the worker count, so pick it for the machine.
35//! - Run [`RenderSession::start_all`] before [`RenderSession::render_all`].
36//!   A plug-in that cannot start otherwise fails every job after two start
37//!   time-outs each, every job returns the sentinel, and a search spends
38//!   its budget on nothing.
39//! - A job that always hangs costs about two render time-outs: one per
40//!   try.
41
42use std::num::NonZeroUsize;
43use std::panic::{AssertUnwindSafe, catch_unwind};
44use std::sync::Mutex;
45use std::sync::atomic::{AtomicUsize, Ordering};
46
47use super::{RenderError, RenderJob, Rendered, Renderer};
48
49/// How many renders a renderer plays before the worker starts a new one.
50pub const RESTART_AFTER: u64 = 256;
51/// How many times a failed render is tried again.
52pub const RETRIES: usize = 1;
53/// The score a consumer gives a failed render: far above any distance a
54/// render can score, so a search ranks it last, and finite, so sums and
55/// means of scores stay finite.
56pub const FAILED_SCORE: f64 = 1.0e9;
57
58/// A renderer a worker can own on its own thread.
59pub type BoxedRenderer = Box<dyn Renderer + Send>;
60
61/// Starts a renderer for the worker with the given index.
62pub type Spawn = dyn Fn(usize) -> Result<BoxedRenderer, RenderError> + Send + Sync;
63
64/// A render that failed on every try.
65#[derive(Debug)]
66pub struct RenderFailure {
67    /// The error of each try, the first first.
68    pub errors: Vec<RenderError>,
69}
70
71impl RenderFailure {
72    /// The score of the failed render, [`FAILED_SCORE`].
73    #[must_use]
74    pub fn score(&self) -> f64 {
75        FAILED_SCORE
76    }
77}
78
79/// What the session gives for one job.
80#[derive(Debug)]
81pub enum RenderOutcome {
82    /// The render.
83    Rendered(Rendered),
84    /// Every try failed.
85    Failed(RenderFailure),
86}
87
88impl RenderOutcome {
89    /// Returns the render, if one came.
90    #[must_use]
91    pub fn rendered(&self) -> Option<&Rendered> {
92        match self {
93            Self::Rendered(rendered) => Some(rendered),
94            Self::Failed(_) => None,
95        }
96    }
97}
98
99/// One worker: its renderer, if one runs, and its counts.
100#[derive(Default)]
101struct Worker {
102    renderer: Option<BoxedRenderer>,
103    /// Renders the current renderer played, failed ones included.
104    renders: u64,
105    /// Renderers this worker started.
106    starts: u64,
107}
108
109/// A pool of workers that renders jobs in parallel.
110pub struct RenderSession {
111    spawn: Box<Spawn>,
112    workers: Vec<Mutex<Worker>>,
113}
114
115impl std::fmt::Debug for RenderSession {
116    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
117        f.debug_struct("RenderSession")
118            .field("workers", &self.workers.len())
119            .finish_non_exhaustive()
120    }
121}
122
123impl RenderSession {
124    /// Builds a session of `workers` workers. `spawn` starts the renderer of
125    /// a worker; it runs on the worker's thread.
126    #[must_use]
127    pub fn new(
128        workers: NonZeroUsize,
129        spawn: impl Fn(usize) -> Result<BoxedRenderer, RenderError> + Send + Sync + 'static,
130    ) -> Self {
131        Self {
132            spawn: Box::new(spawn),
133            workers: (0..workers.get())
134                .map(|_| Mutex::new(Worker::default()))
135                .collect(),
136        }
137    }
138
139    /// Returns how many renderers each worker started so far.
140    #[must_use]
141    pub fn starts(&self) -> Vec<u64> {
142        self.workers
143            .iter()
144            .map(|worker| lock(worker).starts)
145            .collect()
146    }
147
148    /// Starts the renderer of every worker that has none, so a later
149    /// [`RenderSession::render_all`] does not wait for plug-in loads.
150    ///
151    /// # Errors
152    ///
153    /// Returns the first error of a renderer that does not start.
154    pub fn start_all(&self) -> Result<(), RenderError> {
155        std::thread::scope(|scope| {
156            let handles: Vec<_> = self
157                .workers
158                .iter()
159                .enumerate()
160                .map(|(index, worker)| {
161                    scope.spawn(move || {
162                        let mut worker = lock(worker);
163                        if worker.renderer.is_none() {
164                            self.start(index, &mut worker, None)?;
165                        }
166                        Ok(())
167                    })
168                })
169                .collect();
170            handles
171                .into_iter()
172                .try_for_each(|handle| handle.join().unwrap_or_else(|_| Err(panicked())))
173        })
174    }
175
176    /// Renders every job and returns the outcomes in the order of `jobs`.
177    /// The workers take the jobs in turn, each on its own thread.
178    #[must_use]
179    pub fn render_all(&self, jobs: &[RenderJob]) -> Vec<RenderOutcome> {
180        let next = AtomicUsize::new(0);
181        let threads = self.workers.len().min(jobs.len());
182        let mut done: Vec<(usize, RenderOutcome)> = std::thread::scope(|scope| {
183            let handles: Vec<_> = self.workers[..threads]
184                .iter()
185                .enumerate()
186                .map(|(index, worker)| {
187                    let next = &next;
188                    scope.spawn(move || {
189                        let mut worker = lock(worker);
190                        let mut done = Vec::new();
191                        loop {
192                            let at = next.fetch_add(1, Ordering::Relaxed);
193                            let Some(job) = jobs.get(at) else {
194                                break done;
195                            };
196                            done.push((at, self.render_one(index, &mut worker, at, job)));
197                        }
198                    })
199                })
200                .collect();
201            handles
202                .into_iter()
203                .flat_map(|handle| handle.join().unwrap_or_default())
204                .collect()
205        });
206        done.sort_by_key(|(at, _)| *at);
207        // A worker thread that panicked lost its outcomes; its jobs fail.
208        let mut done = done.into_iter().peekable();
209        (0..jobs.len())
210            .map(|at| match done.next_if(|(index, _)| *index == at) {
211                Some((_, outcome)) => outcome,
212                None => RenderOutcome::Failed(RenderFailure {
213                    errors: vec![panicked()],
214                }),
215            })
216            .collect()
217    }
218
219    /// Renders job `at` on `worker`, with a new renderer after
220    /// [`RESTART_AFTER`] renders and after each failure.
221    fn render_one(
222        &self,
223        index: usize,
224        worker: &mut Worker,
225        at: usize,
226        job: &RenderJob,
227    ) -> RenderOutcome {
228        let mut errors = Vec::new();
229        for _ in 0..=RETRIES {
230            if worker.renders >= RESTART_AFTER {
231                worker.renderer = None;
232            }
233            if worker.renderer.is_none()
234                && let Err(error) = self.start(index, worker, Some(at))
235            {
236                errors.push(error);
237                continue;
238            }
239            let Some(renderer) = worker.renderer.as_mut() else {
240                continue;
241            };
242            worker.renders += 1;
243            let result = catch_unwind(AssertUnwindSafe(|| renderer.render(job)))
244                .unwrap_or_else(|payload| Err(panic_error(Some(at), payload.as_ref())));
245            match result {
246                Ok(clip) => return RenderOutcome::Rendered(clip),
247                Err(error @ RenderError::WrongSynth { .. }) => {
248                    // The job is wrong, not the plug-in: a retry gives the
249                    // same error, and the renderer stays healthy.
250                    errors.push(error);
251                    break;
252                }
253                Err(error) => {
254                    errors.push(error);
255                    // A failed plug-in may hold a broken state, so the try
256                    // after a failure runs on a new renderer.
257                    worker.renderer = None;
258                }
259            }
260        }
261        RenderOutcome::Failed(RenderFailure { errors })
262    }
263
264    fn start(
265        &self,
266        index: usize,
267        worker: &mut Worker,
268        at: Option<usize>,
269    ) -> Result<(), RenderError> {
270        // The old renderer ends before the new one starts, so a worker
271        // never holds two plug-in instances.
272        worker.renderer = None;
273        worker.starts += 1;
274        worker.renders = 0;
275        let started = catch_unwind(AssertUnwindSafe(|| (self.spawn)(index)))
276            .unwrap_or_else(|payload| Err(panic_error(at, payload.as_ref())))?;
277        worker.renderer = Some(started);
278        Ok(())
279    }
280}
281
282/// Locks a worker. A worker whose thread panicked is still usable: its
283/// renderer is dropped, so the next job starts a new one.
284fn lock(worker: &Mutex<Worker>) -> std::sync::MutexGuard<'_, Worker> {
285    worker.lock().unwrap_or_else(|poisoned| {
286        let mut worker = poisoned.into_inner();
287        worker.renderer = None;
288        worker
289    })
290}
291
292fn panicked() -> RenderError {
293    RenderError::Internal("a render worker thread panicked".to_owned())
294}
295
296/// The error of a renderer that panicked, naming the job it worked on.
297fn panic_error(at: Option<usize>, payload: &(dyn std::any::Any + Send)) -> RenderError {
298    let message = payload
299        .downcast_ref::<&str>()
300        .map(|text| (*text).to_owned())
301        .or_else(|| payload.downcast_ref::<String>().cloned())
302        .unwrap_or_else(|| "no message".to_owned());
303    let job = at.map_or_else(
304        || "while it started".to_owned(),
305        |at| format!("on job {at}"),
306    );
307    RenderError::Internal(format!("the renderer panicked {job}: {message}"))
308}
309
310#[cfg(test)]
311mod tests {
312    use super::*;
313
314    #[test]
315    fn a_poisoned_worker_drops_its_renderer() {
316        let worker = Mutex::new(Worker {
317            renderer: Some(Box::new(crate::render::fake::FakeRenderer::new(
318                crate::synth::Synth::Vital,
319                0,
320            ))),
321            renders: 3,
322            starts: 1,
323        });
324        let poisoned = std::thread::scope(|scope| {
325            scope
326                .spawn(|| {
327                    let _guard = worker.lock().unwrap();
328                    panic!("poison the lock");
329                })
330                .join()
331                .is_err()
332        });
333        assert!(poisoned && worker.is_poisoned());
334        let worker = lock(&worker);
335        assert!(worker.renderer.is_none());
336        assert_eq!((worker.renders, worker.starts), (3, 1));
337    }
338
339    #[test]
340    fn a_panic_names_the_job_and_its_message() {
341        let error = panic_error(Some(6), &"boom");
342        let message = error.to_string();
343        assert!(
344            message.contains("the renderer panicked on job 6: boom"),
345            "{message}"
346        );
347        assert!(message.contains("fault of this program"), "{message}");
348        assert!(!message.contains("plug-in host"), "{message}");
349        let error = panic_error(None, &String::from("bad"));
350        assert!(error.to_string().contains("panicked while it started: bad"));
351    }
352}
sound_to_patch/render/session.rs

sound_to_patch/render/
session.rs
