rustc_data_structures/sync/
parallel.rs

1//! This module defines parallel operations that are implemented in
2//! one way for the serial compiler, and another way the parallel compiler.
3
4use std::any::Any;
5use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind};
6
7use parking_lot::Mutex;
8
9use crate::FatalErrorMarker;
10use crate::sync::{DynSend, DynSync, FromDyn, IntoDynSyncSend, mode};
11
12/// A guard used to hold panics that occur during a parallel section to later by unwound.
13/// This is used for the parallel compiler to prevent fatal errors from non-deterministically
14/// hiding errors by ensuring that everything in the section has completed executing before
15/// continuing with unwinding. It's also used for the non-parallel code to ensure error message
16/// output match the parallel compiler for testing purposes.
17pub struct ParallelGuard {
18    panic: Mutex<Option<IntoDynSyncSend<Box<dyn Any + Send + 'static>>>>,
19}
20
21impl ParallelGuard {
22    pub fn run<R>(&self, f: impl FnOnce() -> R) -> Option<R> {
23        catch_unwind(AssertUnwindSafe(f))
24            .map_err(|err| {
25                let mut panic = self.panic.lock();
26                if panic.is_none() || !(*err).is::<FatalErrorMarker>() {
27                    *panic = Some(IntoDynSyncSend(err));
28                }
29            })
30            .ok()
31    }
32}
33
34/// This gives access to a fresh parallel guard in the closure and will unwind any panics
35/// caught in it after the closure returns.
36#[inline]
37pub fn parallel_guard<R>(f: impl FnOnce(&ParallelGuard) -> R) -> R {
38    let guard = ParallelGuard { panic: Mutex::new(None) };
39    let ret = f(&guard);
40    if let Some(IntoDynSyncSend(panic)) = guard.panic.into_inner() {
41        resume_unwind(panic);
42    }
43    ret
44}
45
46fn serial_join<A, B, RA, RB>(oper_a: A, oper_b: B) -> (RA, RB)
47where
48    A: FnOnce() -> RA,
49    B: FnOnce() -> RB,
50{
51    let (a, b) = parallel_guard(|guard| {
52        let a = guard.run(oper_a);
53        let b = guard.run(oper_b);
54        (a, b)
55    });
56    (a.unwrap(), b.unwrap())
57}
58
59pub fn spawn(func: impl FnOnce() + DynSend + 'static) {
60    if mode::is_dyn_thread_safe() {
61        let func = FromDyn::from(func);
62        rustc_thread_pool::spawn(|| {
63            (func.into_inner())();
64        });
65    } else {
66        func()
67    }
68}
69
70/// Runs the functions in parallel.
71///
72/// The first function is executed immediately on the current thread.
73/// Use that for the longest running function for better scheduling.
74pub fn par_fns(funcs: &mut [&mut (dyn FnMut() + DynSend)]) {
75    parallel_guard(|guard: &ParallelGuard| {
76        if mode::is_dyn_thread_safe() {
77            let funcs = FromDyn::from(funcs);
78            rustc_thread_pool::scope(|s| {
79                let Some((first, rest)) = funcs.into_inner().split_at_mut_checked(1) else {
80                    return;
81                };
82
83                // Reverse the order of the later functions since Rayon executes them in reverse
84                // order when using a single thread. This ensures the execution order matches
85                // that of a single threaded rustc.
86                for f in rest.iter_mut().rev() {
87                    let f = FromDyn::from(f);
88                    s.spawn(|_| {
89                        guard.run(|| (f.into_inner())());
90                    });
91                }
92
93                // Run the first function without spawning to
94                // ensure it executes immediately on this thread.
95                guard.run(|| first[0]());
96            });
97        } else {
98            for f in funcs {
99                guard.run(|| f());
100            }
101        }
102    });
103}
104
105#[inline]
106pub fn par_join<A, B, RA: DynSend, RB: DynSend>(oper_a: A, oper_b: B) -> (RA, RB)
107where
108    A: FnOnce() -> RA + DynSend,
109    B: FnOnce() -> RB + DynSend,
110{
111    if mode::is_dyn_thread_safe() {
112        let oper_a = FromDyn::from(oper_a);
113        let oper_b = FromDyn::from(oper_b);
114        let (a, b) = parallel_guard(|guard| {
115            rustc_thread_pool::join(
116                move || guard.run(move || FromDyn::from(oper_a.into_inner()())),
117                move || guard.run(move || FromDyn::from(oper_b.into_inner()())),
118            )
119        });
120        (a.unwrap().into_inner(), b.unwrap().into_inner())
121    } else {
122        serial_join(oper_a, oper_b)
123    }
124}
125
126fn par_slice<I: DynSend>(
127    items: &mut [I],
128    guard: &ParallelGuard,
129    for_each: impl Fn(&mut I) + DynSync + DynSend,
130) {
131    struct State<'a, F> {
132        for_each: FromDyn<F>,
133        guard: &'a ParallelGuard,
134        group: usize,
135    }
136
137    fn par_rec<I: DynSend, F: Fn(&mut I) + DynSync + DynSend>(
138        items: &mut [I],
139        state: &State<'_, F>,
140    ) {
141        if items.len() <= state.group {
142            for item in items {
143                state.guard.run(|| (state.for_each)(item));
144            }
145        } else {
146            let (left, right) = items.split_at_mut(items.len() / 2);
147            let mut left = state.for_each.derive(left);
148            let mut right = state.for_each.derive(right);
149            rustc_thread_pool::join(move || par_rec(*left, state), move || par_rec(*right, state));
150        }
151    }
152
153    let state = State {
154        for_each: FromDyn::from(for_each),
155        guard,
156        group: std::cmp::max(items.len() / 128, 1),
157    };
158    par_rec(items, &state)
159}
160
161pub fn par_for_each_in<I: DynSend, T: IntoIterator<Item = I>>(
162    t: T,
163    for_each: impl Fn(&I) + DynSync + DynSend,
164) {
165    parallel_guard(|guard| {
166        if mode::is_dyn_thread_safe() {
167            let mut items: Vec<_> = t.into_iter().collect();
168            par_slice(&mut items, guard, |i| for_each(&*i))
169        } else {
170            t.into_iter().for_each(|i| {
171                guard.run(|| for_each(&i));
172            });
173        }
174    });
175}
176
177/// This runs `for_each` in parallel for each iterator item. If one or more of the
178/// `for_each` calls returns `Err`, the function will also return `Err`. The error returned
179/// will be non-deterministic, but this is expected to be used with `ErrorGuaranteed` which
180/// are all equivalent.
181pub fn try_par_for_each_in<T: IntoIterator, E: DynSend>(
182    t: T,
183    for_each: impl Fn(&<T as IntoIterator>::Item) -> Result<(), E> + DynSync + DynSend,
184) -> Result<(), E>
185where
186    <T as IntoIterator>::Item: DynSend,
187{
188    parallel_guard(|guard| {
189        if mode::is_dyn_thread_safe() {
190            let mut items: Vec<_> = t.into_iter().collect();
191
192            let error = Mutex::new(None);
193
194            par_slice(&mut items, guard, |i| {
195                if let Err(err) = for_each(&*i) {
196                    *error.lock() = Some(err);
197                }
198            });
199
200            if let Some(err) = error.into_inner() { Err(err) } else { Ok(()) }
201        } else {
202            t.into_iter().filter_map(|i| guard.run(|| for_each(&i))).fold(Ok(()), Result::and)
203        }
204    })
205}
206
207pub fn par_map<I: DynSend, T: IntoIterator<Item = I>, R: DynSend, C: FromIterator<R>>(
208    t: T,
209    map: impl Fn(I) -> R + DynSync + DynSend,
210) -> C {
211    parallel_guard(|guard| {
212        if mode::is_dyn_thread_safe() {
213            let map = FromDyn::from(map);
214
215            let mut items: Vec<(Option<I>, Option<R>)> =
216                t.into_iter().map(|i| (Some(i), None)).collect();
217
218            par_slice(&mut items, guard, |i| {
219                i.1 = Some(map(i.0.take().unwrap()));
220            });
221
222            items.into_iter().filter_map(|i| i.1).collect()
223        } else {
224            t.into_iter().filter_map(|i| guard.run(|| map(i))).collect()
225        }
226    })
227}
228
229pub fn broadcast<R: DynSend>(op: impl Fn(usize) -> R + DynSync) -> Vec<R> {
230    if mode::is_dyn_thread_safe() {
231        let op = FromDyn::from(op);
232        let results = rustc_thread_pool::broadcast(|context| op.derive(op(context.index())));
233        results.into_iter().map(|r| r.into_inner()).collect()
234    } else {
235        <[_]>::into_vec(::alloc::boxed::box_new([op(0)]))vec![op(0)]
236    }
237}
rustc_data_structures/sync/parallel.rs

rustc_data_structures/sync/
parallel.rs
