Lower thread::spawn by closure conversion, and emit the build flag it needs
Rust's thread::spawn takes a closure that captures. A Nim thread proc is
{.thread, nimcall.} and deliberately cannot, since a closure refers to the
creating thread's environment. So each spawn site is closure-converted: the
body becomes a top-level {.thread.} proc and its captures become a heap
object passed as the argument, which also carries the result because a Nim
thread proc returns nothing. `move` is what Rust requires for spawn and
copying the captures is exactly what it means, so the move closure that is
rejected everywhere else is the point here.
JoinHandle<T> is uniform in T: the per-site capture object reaches the thread
as an opaque pointer, kept alive by the handle. The first design gave each
spawn site its own handle type, which no Rust source could name -- so
Vec<JoinHandle<T>> did not typecheck.
The Arc caveat came due. When Arc was added it was recorded as a difference
that "matters only across threads": Nim's reference counting is not atomic.
Adding threads made it matter, and the failure was a segfault inside ORC's
cycle collector with stdout already fully correct -- which is the worst shape
a bug can have. Nim's answer is --mm:atomicArc, a compiler setting that no
source can express, so rustnim now writes a companion <output>.nim.cfg when
the program spawns threads. Nim reads that automatically, so the build
requirement travels with the output instead of being a line in a README. A
program without threads gets no config, and a stale one is removed rather
than left to change the next build silently.
A panicking thread is not isolated: Rust's join returns Err and the process
carries on, where panics here map to quit(101) and end everything. Recorded
rather than papered over.
tests/cases/046 spawns four threads incrementing shared state through both
Arc<Mutex<i32>> and Arc<AtomicUsize>. The totals are deterministic though the
interleaving is not, which is what makes it a differential test rather than a
coin flip.
52 differential cases, all green.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>1b93db2 parent: a721ac2 modified
DESIGN.md +42 -0 | @@ -901,6 +901,48 @@ Two things needed care: | ||
| 901 | 901 | its `expected` argument. The prelude bridges them, and the test pins both |
| 902 | 902 | outcomes. |
| 903 | 903 | |
| 904 | +### Threads | |
| 905 | + | |
| 906 | +Rust's `thread::spawn` takes a closure that captures. A Nim thread proc is | |
| 907 | +`{.thread, nimcall.}` and deliberately **cannot** — a closure refers to the | |
| 908 | +creating thread's environment. So each spawn site is **closure-converted**: | |
| 909 | +the body becomes a top-level `{.thread.}` proc, and whatever it captured | |
| 910 | +becomes a heap object passed as the argument, which also carries the result | |
| 911 | +because a Nim thread proc returns nothing. `move` is what Rust requires for | |
| 912 | +`spawn`, and copying the captures into that object is exactly what `move` | |
| 913 | +means — so the `move` closure that is rejected everywhere else is the whole | |
| 914 | +point here. | |
| 915 | + | |
| 916 | +`JoinHandle<T>` is uniform in `T`: the per-site capture object reaches the | |
| 917 | +thread as an opaque pointer, kept alive by the handle. Without that, handles | |
| 918 | +from one spawn site would have a Nim type no Rust source could name, and | |
| 919 | +`Vec<JoinHandle<T>>` would not typecheck. | |
| 920 | + | |
| 921 | +`tests/cases/046` spawns four threads incrementing shared state through both | |
| 922 | +`Arc<Mutex<i32>>` and `Arc<AtomicUsize>`. The totals are deterministic even | |
| 923 | +though the interleaving is not, which is what makes it a differential test | |
| 924 | +rather than a coin flip. | |
| 925 | + | |
| 926 | +#### The `Arc` caveat came due | |
| 927 | + | |
| 928 | +When `Arc` was added, this was recorded as a difference that "matters only | |
| 929 | +across threads": Nim's reference counting is **not atomic**. Adding threads | |
| 930 | +made it matter, and the failure was not subtle — a segfault inside ORC's cycle | |
| 931 | +collector, with stdout already fully correct. | |
| 932 | + | |
| 933 | +Nim's answer is `--mm:atomicArc`, which is a *compiler setting* and not | |
| 934 | +anything the source can express. So rustnim writes a companion | |
| 935 | +`<output>.nim.cfg` when the program spawns threads. Nim reads that file | |
| 936 | +automatically, which means the build requirement travels with the output | |
| 937 | +instead of being a line in a README that someone has to read. A program | |
| 938 | +without threads gets no config, and a stale one from an earlier run is | |
| 939 | +removed rather than left to silently change how the next build works. | |
| 940 | + | |
| 941 | +**A panicking thread is not isolated.** Rust's `join` returns `Err` when the | |
| 942 | +thread panicked and the process carries on; panics here map to `quit(101)`, | |
| 943 | +which ends everything. A program relying on catching a thread panic would | |
| 944 | +differ. | |
| 945 | + | |
| 904 | 946 | ### The prelude is composed last |
| 905 | 947 | |
| 906 | 948 | Whether a program needs the async dispatcher or the locking runtime is |
| @@ -901,6 +901,48 @@ Two things needed care: | |||
| 901 | its `expected` argument. The prelude bridges them, and the test pins both | 901 | its `expected` argument. The prelude bridges them, and the test pins both |
| 902 | outcomes. | 902 | outcomes. |
| 903 | 903 | ||
| 904 | +### Threads | ||
| 905 | + | ||
| 906 | +Rust's `thread::spawn` takes a closure that captures. A Nim thread proc is | ||
| 907 | +`{.thread, nimcall.}` and deliberately **cannot** — a closure refers to the | ||
| 908 | +creating thread's environment. So each spawn site is **closure-converted**: | ||
| 909 | +the body becomes a top-level `{.thread.}` proc, and whatever it captured | ||
| 910 | +becomes a heap object passed as the argument, which also carries the result | ||
| 911 | +because a Nim thread proc returns nothing. `move` is what Rust requires for | ||
| 912 | +`spawn`, and copying the captures into that object is exactly what `move` | ||
| 913 | +means — so the `move` closure that is rejected everywhere else is the whole | ||
| 914 | +point here. | ||
| 915 | + | ||
| 916 | +`JoinHandle<T>` is uniform in `T`: the per-site capture object reaches the | ||
| 917 | +thread as an opaque pointer, kept alive by the handle. Without that, handles | ||
| 918 | +from one spawn site would have a Nim type no Rust source could name, and | ||
| 919 | +`Vec<JoinHandle<T>>` would not typecheck. | ||
| 920 | + | ||
| 921 | +`tests/cases/046` spawns four threads incrementing shared state through both | ||
| 922 | +`Arc<Mutex<i32>>` and `Arc<AtomicUsize>`. The totals are deterministic even | ||
| 923 | +though the interleaving is not, which is what makes it a differential test | ||
| 924 | +rather than a coin flip. | ||
| 925 | + | ||
| 926 | +#### The `Arc` caveat came due | ||
| 927 | + | ||
| 928 | +When `Arc` was added, this was recorded as a difference that "matters only | ||
| 929 | +across threads": Nim's reference counting is **not atomic**. Adding threads | ||
| 930 | +made it matter, and the failure was not subtle — a segfault inside ORC's cycle | ||
| 931 | +collector, with stdout already fully correct. | ||
| 932 | + | ||
| 933 | +Nim's answer is `--mm:atomicArc`, which is a *compiler setting* and not | ||
| 934 | +anything the source can express. So rustnim writes a companion | ||
| 935 | +`<output>.nim.cfg` when the program spawns threads. Nim reads that file | ||
| 936 | +automatically, which means the build requirement travels with the output | ||
| 937 | +instead of being a line in a README that someone has to read. A program | ||
| 938 | +without threads gets no config, and a stale one from an earlier run is | ||
| 939 | +removed rather than left to silently change how the next build works. | ||
| 940 | + | ||
| 941 | +**A panicking thread is not isolated.** Rust's `join` returns `Err` when the | ||
| 942 | +thread panicked and the process carries on; panics here map to `quit(101)`, | ||
| 943 | +which ends everything. A program relying on catching a thread panic would | ||
| 944 | +differ. | ||
| 945 | + | ||
| 904 | ### The prelude is composed last | 946 | ### The prelude is composed last |
| 905 | 947 | ||
| 906 | Whether a program needs the async dispatcher or the locking runtime is | 948 | Whether a program needs the async dispatcher or the locking runtime is |
modified
README.md +8 -0 | @@ -136,6 +136,14 @@ the obvious mapping reorders every side effect ahead of the first await. | ||
| 136 | 136 | Measured side by side, then fixed by making a future |
| 137 | 137 | `proc (): Future[T]`, which `.await` calls. |
| 138 | 138 | |
| 139 | +## Concurrency | |
| 140 | + | |
| 141 | +`Arc`, `Mutex`, atomics and `thread::spawn` all lower. A spawn site is | |
| 142 | +closure-converted, because Nim's thread procs cannot capture. Sharing | |
| 143 | +reference-counted values across threads needs atomic refcounting, so rustnim | |
| 144 | +writes a `<output>.nim.cfg` next to the output asking for it — Nim reads that | |
| 145 | +automatically, so the requirement travels with the code. | |
| 146 | + | |
| 139 | 147 | ## Does it generalise? |
| 140 | 148 | |
| 141 | 149 | `base16ct` is the crate this was built toward, so a second one was tried. |
| @@ -136,6 +136,14 @@ the obvious mapping reorders every side effect ahead of the first await. | |||
| 136 | Measured side by side, then fixed by making a future | 136 | Measured side by side, then fixed by making a future |
| 137 | `proc (): Future[T]`, which `.await` calls. | 137 | `proc (): Future[T]`, which `.await` calls. |
| 138 | 138 | ||
| 139 | +## Concurrency | ||
| 140 | + | ||
| 141 | +`Arc`, `Mutex`, atomics and `thread::spawn` all lower. A spawn site is | ||
| 142 | +closure-converted, because Nim's thread procs cannot capture. Sharing | ||
| 143 | +reference-counted values across threads needs atomic refcounting, so rustnim | ||
| 144 | +writes a `<output>.nim.cfg` next to the output asking for it — Nim reads that | ||
| 145 | +automatically, so the requirement travels with the code. | ||
| 146 | + | ||
| 139 | ## Does it generalise? | 147 | ## Does it generalise? |
| 140 | 148 | ||
| 141 | `base16ct` is the crate this was built toward, so a second one was tried. | 149 | `base16ct` is the crate this was built toward, so a second one was tried. |
modified
src/lower.rs +168 -1 | @@ -279,6 +279,16 @@ pub struct Lowerer { | ||
| 279 | 279 | aliases: HashMap<String, (Vec<String>, syn::Type)>, |
| 280 | 280 | /// Whether anything in the input is a future, which decides if the async |
| 281 | 281 | /// runtime is emitted. |
| 282 | + /// Whether the program spawns threads. Sharing a `ref` across threads | |
| 283 | + /// needs atomic reference counting, which is a compiler setting rather | |
| 284 | + /// than anything expressible in the source. | |
| 285 | + pub uses_threads: bool, | |
| 286 | + /// Items that must be emitted at top level though they are discovered | |
| 287 | + /// inside a function -- a thread body and its capture object. | |
| 288 | + hoisted: String, | |
| 289 | + /// Where the bodies begin in `out`, so hoisted items can be spliced in | |
| 290 | + /// before them. | |
| 291 | + body_start: usize, | |
| 282 | 292 | /// Locks acquired while lowering the current statement, awaiting a |
| 283 | 293 | /// decision about where they are released. |
| 284 | 294 | pending_locks: Vec<String>, |
| @@ -347,6 +357,9 @@ impl Lowerer { | ||
| 347 | 357 | vec_expect: None, |
| 348 | 358 | forwards: Vec::new(), |
| 349 | 359 | aliases: HashMap::new(), |
| 360 | + uses_threads: false, | |
| 361 | + hoisted: String::new(), | |
| 362 | + body_start: 0, | |
| 350 | 363 | pending_locks: Vec::new(), |
| 351 | 364 | uses_async: false, |
| 352 | 365 | uses_sync: std::cell::Cell::new(false), |
| @@ -449,6 +462,8 @@ impl Lowerer { | ||
| 449 | 462 | self.blank(); |
| 450 | 463 | } |
| 451 | 464 | |
| 465 | + self.body_start = self.out.len(); | |
| 466 | + | |
| 452 | 467 | // Pass 4: bodies. |
| 453 | 468 | for (m, f) in files { |
| 454 | 469 | self.cur_mod = m.clone(); |
| @@ -498,7 +513,13 @@ impl Lowerer { | ||
| 498 | 513 | // is needed can be discovered while lowering rather than predicted |
| 499 | 514 | // before it. The async dispatcher and the locking runtime each cost |
| 500 | 515 | // compile time, and most programs need neither. |
| 501 | - let body = std::mem::take(&mut self.out); | |
| 516 | + // Hoisted items go between the declarations and the bodies: they are | |
| 517 | + // top-level procs discovered while lowering a function. | |
| 518 | + let mut body = std::mem::take(&mut self.out); | |
| 519 | + if !self.hoisted.is_empty() { | |
| 520 | + let at = self.body_start.min(body.len()); | |
| 521 | + body.insert_str(at, &std::mem::take(&mut self.hoisted)); | |
| 522 | + } | |
| 502 | 523 | let mut out = String::from(include_str!("prelude.nim")); |
| 503 | 524 | out.push('\n'); |
| 504 | 525 | if self.uses_async { |
| @@ -3122,6 +3143,109 @@ impl Lowerer { | ||
| 3122 | 3143 | Ok(()) |
| 3123 | 3144 | } |
| 3124 | 3145 | |
| 3146 | + /// Lower `thread::spawn(move || body)`. | |
| 3147 | + /// | |
| 3148 | + /// The body becomes a top-level `{.thread.}` proc and its captures become | |
| 3149 | + /// a heap object passed as the argument -- which also carries the result, | |
| 3150 | + /// since a Nim thread proc returns nothing. | |
| 3151 | + fn spawn_thread(&mut self, cl: &syn::ExprClosure) -> Result<Val, String> { | |
| 3152 | + if cl.capture.is_none() { | |
| 3153 | + return Err("`thread::spawn` needs a `move` closure: the thread \ | |
| 3154 | + outlives this scope, so its captures must be owned" | |
| 3155 | + .into()); | |
| 3156 | + } | |
| 3157 | + if !cl.inputs.is_empty() { | |
| 3158 | + return Err("`thread::spawn` takes a closure with no arguments".into()); | |
| 3159 | + } | |
| 3160 | + self.uses_sync.set(true); | |
| 3161 | + self.uses_threads = true; | |
| 3162 | + | |
| 3163 | + // Whatever the body names that is bound out here is a capture. | |
| 3164 | + let mut names = Vec::new(); | |
| 3165 | + collect_idents(&cl.body, &mut names); | |
| 3166 | + let mut caps: Vec<(String, Nim)> = Vec::new(); | |
| 3167 | + for n in names { | |
| 3168 | + if caps.iter().any(|(m, _)| *m == n) { | |
| 3169 | + continue; | |
| 3170 | + } | |
| 3171 | + if let Some(t) = self.lookup(&n) { | |
| 3172 | + caps.push((n, t)); | |
| 3173 | + } else if let Some(Alias::Value { ty: Some(t), .. }) = self.lookup_alias(&n) { | |
| 3174 | + caps.push((n, t)); | |
| 3175 | + } | |
| 3176 | + } | |
| 3177 | + | |
| 3178 | + let id = self.fresh("Spawn"); | |
| 3179 | + let state = format!("{}State", id); | |
| 3180 | + let body = match &*cl.body { | |
| 3181 | + Expr::Block(b) => b.block.clone(), | |
| 3182 | + other => syn::Block { | |
| 3183 | + brace_token: Default::default(), | |
| 3184 | + stmts: vec![Stmt::Expr(other.clone(), None)], | |
| 3185 | + }, | |
| 3186 | + }; | |
| 3187 | + | |
| 3188 | + // Lower the body against the capture object, into the hoisted buffer. | |
| 3189 | + let outer_out = std::mem::take(&mut self.out); | |
| 3190 | + let outer_indent = std::mem::replace(&mut self.indent, 0); | |
| 3191 | + self.push_scope(); | |
| 3192 | + for (n, t) in &caps { | |
| 3193 | + self.bind_alias( | |
| 3194 | + n, | |
| 3195 | + Alias::Value { code: format!("a.{}", ident(n)), ty: Some(t.clone()) }, | |
| 3196 | + ); | |
| 3197 | + } | |
| 3198 | + // The thread takes an opaque pointer so that every handle is the same | |
| 3199 | + // Nim type; the capture object is recovered inside. | |
| 3200 | + self.line(&format!("proc rsThread{}(p: pointer) {{.thread.}} =", id)); | |
| 3201 | + self.indent += 1; | |
| 3202 | + self.line(&format!("let a = cast[{}](p)", state)); | |
| 3203 | + let outer_t = self.target.replace(("a.res[]".to_string(), None)); | |
| 3204 | + let before = self.out.len(); | |
| 3205 | + let tail = self.block_body(&body)?; | |
| 3206 | + let out_ty = tail.as_ref().and_then(|v| v.ty.clone()); | |
| 3207 | + self.emit_tail(tail); | |
| 3208 | + if self.out.len() == before { | |
| 3209 | + self.line("discard"); | |
| 3210 | + } | |
| 3211 | + self.target = outer_t; | |
| 3212 | + self.pop_scope(); | |
| 3213 | + self.indent = outer_indent; | |
| 3214 | + let proc_text = std::mem::replace(&mut self.out, outer_out); | |
| 3215 | + | |
| 3216 | + // The state type has to be declared before the proc that takes it. | |
| 3217 | + let res_ty = out_ty.clone().unwrap_or(Nim::Unit); | |
| 3218 | + let mut decl = format!("type {}* = ref object of RootObj\n", state); | |
| 3219 | + for (n, t) in &caps { | |
| 3220 | + decl.push_str(&format!(" {}*: {}\n", ident(n), t.render())); | |
| 3221 | + } | |
| 3222 | + decl.push_str(&format!(" res*: ref {}\n\n", res_ty.render())); | |
| 3223 | + self.hoisted.push_str(&decl); | |
| 3224 | + self.hoisted.push_str(&proc_text); | |
| 3225 | + self.hoisted.push('\n'); | |
| 3226 | + | |
| 3227 | + // Build the state, then start the thread. | |
| 3228 | + let st = self.fresh("St"); | |
| 3229 | + let h = self.fresh("H"); | |
| 3230 | + self.line(&format!("var {}: {}", st, state)); | |
| 3231 | + self.line(&format!("new({})", st)); | |
| 3232 | + for (n, _) in &caps { | |
| 3233 | + let v = match self.lookup_alias(n) { | |
| 3234 | + Some(Alias::Value { code, .. }) => code, | |
| 3235 | + _ => ident(n), | |
| 3236 | + }; | |
| 3237 | + self.line(&format!("{}.{} = {}", st, ident(n), v)); | |
| 3238 | + } | |
| 3239 | + self.line(&format!("let {} = rsNewJoinHandle[{}]()", h, res_ty.render())); | |
| 3240 | + self.line(&format!("{}.res = {}.res", st, h)); | |
| 3241 | + self.line(&format!("{}.st = RootRef({})", h, st)); | |
| 3242 | + self.line(&format!( | |
| 3243 | + "createThread({}.th, rsThread{}, cast[pointer]({}))", | |
| 3244 | + h, id, st | |
| 3245 | + )); | |
| 3246 | + Ok(Val::new(h, Some(Nim::Named("RsJoinHandle".into(), vec![res_ty])))) | |
| 3247 | + } | |
| 3248 | + | |
| 3125 | 3249 | /// Bind one element of an iterator under a fresh name, and return how to |
| 3126 | 3250 | /// refer to it and what type it has. |
| 3127 | 3251 | fn bind_element(&mut self, it: &Iter, i: &str) -> Result<(String, Option<Nim>), String> { |
| @@ -4723,6 +4847,16 @@ impl Lowerer { | ||
| 4723 | 4847 | return Err("only calls to named functions are supported".into()); |
| 4724 | 4848 | }; |
| 4725 | 4849 | let name = path_name(&p.path); |
| 4850 | + | |
| 4851 | + // `thread::spawn(move || ..)` is handled before the arguments are | |
| 4852 | + // lowered: a `move` closure is rejected everywhere else, and here it | |
| 4853 | + // is the whole point. See `src/prelude_sync.nim`. | |
| 4854 | + if name == "spawn" && c.args.len() == 1 { | |
| 4855 | + if let Expr::Closure(cl) = &c.args[0] { | |
| 4856 | + return self.spawn_thread(cl); | |
| 4857 | + } | |
| 4858 | + } | |
| 4859 | + | |
| 4726 | 4860 | let target = self.resolve_fn(&p.path); |
| 4727 | 4861 | let ptys: Vec<Nim> = target |
| 4728 | 4862 | .as_ref() |
| @@ -5227,6 +5361,11 @@ impl Lowerer { | ||
| 5227 | 5361 | // which would look for a `Result`. |
| 5228 | 5362 | if matches!(name.as_str(), "unwrap" | "expect") { |
| 5229 | 5363 | if let Some(Nim::Named(n, a)) = &rt { |
| 5364 | + // `join()` already produced the value; Rust wraps it in a | |
| 5365 | + // `Result` only because a panicking thread is caught there. | |
| 5366 | + if n == "RsJoinOut" { | |
| 5367 | + return Ok(Val::new(recv.code.clone(), a.first().cloned())); | |
| 5368 | + } | |
| 5230 | 5369 | if n == "RsGuard" { |
| 5231 | 5370 | let inner = a.first().cloned(); |
| 5232 | 5371 | let mut v = Val::new(recv.code.clone(), inner.clone()); |
| @@ -5458,6 +5597,18 @@ impl Lowerer { | ||
| 5458 | 5597 | Some(Nim::Ptr(Box::new(elem))), |
| 5459 | 5598 | ) |
| 5460 | 5599 | } |
| 5600 | + // `h.join()` waits and yields what the thread produced. Rust's | |
| 5601 | + // returns `Result` because a panicking thread is caught there; | |
| 5602 | + // here a panic ends the process, so there is no error case -- see | |
| 5603 | + // `src/prelude_sync.nim`. | |
| 5604 | + "join" if matches!(&rt, Some(Nim::Named(n, _)) if n == "RsJoinHandle") => { | |
| 5605 | + let Some(Nim::Named(_, a)) = &rt else { unreachable!() }; | |
| 5606 | + let out = a.first().cloned(); | |
| 5607 | + return Ok(Val::new( | |
| 5608 | + format!("rsJoin({})", recv.code), | |
| 5609 | + Some(Nim::Named("RsJoinOut".into(), out.into_iter().collect())), | |
| 5610 | + )); | |
| 5611 | + } | |
| 5461 | 5612 | // Atomic operations. Each is the same C11 primitive on both |
| 5462 | 5613 | // sides, with the ordering carried through unchanged. |
| 5463 | 5614 | "load" | "store" | "swap" | "fetch_add" | "fetch_sub" | "fetch_and" |
| @@ -6108,6 +6259,22 @@ fn module_of(prefix: &[String]) -> String { | ||
| 6108 | 6259 | } |
| 6109 | 6260 | } |
| 6110 | 6261 | |
| 6262 | +/// Every single-segment identifier an expression mentions. Used to find what | |
| 6263 | +/// a `move` closure captures: a name it uses that is bound outside it. | |
| 6264 | +fn collect_idents(e: &Expr, out: &mut Vec<String>) { | |
| 6265 | + use syn::visit::Visit; | |
| 6266 | + struct V<'a>(&'a mut Vec<String>); | |
| 6267 | + impl<'ast> Visit<'ast> for V<'_> { | |
| 6268 | + fn visit_expr_path(&mut self, p: &'ast syn::ExprPath) { | |
| 6269 | + if p.path.segments.len() == 1 && p.qself.is_none() { | |
| 6270 | + self.0.push(p.path.segments[0].ident.to_string()); | |
| 6271 | + } | |
| 6272 | + syn::visit::visit_expr_path(self, p); | |
| 6273 | + } | |
| 6274 | + } | |
| 6275 | + V(out).visit_expr(e); | |
| 6276 | +} | |
| 6277 | + | |
| 6111 | 6278 | /// Whether a type involves the locking runtime. |
| 6112 | 6279 | fn mentions_sync(t: &Nim) -> bool { |
| 6113 | 6280 | match t { |
| @@ -279,6 +279,16 @@ pub struct Lowerer { | |||
| 279 | aliases: HashMap<String, (Vec<String>, syn::Type)>, | 279 | aliases: HashMap<String, (Vec<String>, syn::Type)>, |
| 280 | /// Whether anything in the input is a future, which decides if the async | 280 | /// Whether anything in the input is a future, which decides if the async |
| 281 | /// runtime is emitted. | 281 | /// runtime is emitted. |
| 282 | + /// Whether the program spawns threads. Sharing a `ref` across threads | ||
| 283 | + /// needs atomic reference counting, which is a compiler setting rather | ||
| 284 | + /// than anything expressible in the source. | ||
| 285 | + pub uses_threads: bool, | ||
| 286 | + /// Items that must be emitted at top level though they are discovered | ||
| 287 | + /// inside a function -- a thread body and its capture object. | ||
| 288 | + hoisted: String, | ||
| 289 | + /// Where the bodies begin in `out`, so hoisted items can be spliced in | ||
| 290 | + /// before them. | ||
| 291 | + body_start: usize, | ||
| 282 | /// Locks acquired while lowering the current statement, awaiting a | 292 | /// Locks acquired while lowering the current statement, awaiting a |
| 283 | /// decision about where they are released. | 293 | /// decision about where they are released. |
| 284 | pending_locks: Vec<String>, | 294 | pending_locks: Vec<String>, |
| @@ -347,6 +357,9 @@ impl Lowerer { | |||
| 347 | vec_expect: None, | 357 | vec_expect: None, |
| 348 | forwards: Vec::new(), | 358 | forwards: Vec::new(), |
| 349 | aliases: HashMap::new(), | 359 | aliases: HashMap::new(), |
| 360 | + uses_threads: false, | ||
| 361 | + hoisted: String::new(), | ||
| 362 | + body_start: 0, | ||
| 350 | pending_locks: Vec::new(), | 363 | pending_locks: Vec::new(), |
| 351 | uses_async: false, | 364 | uses_async: false, |
| 352 | uses_sync: std::cell::Cell::new(false), | 365 | uses_sync: std::cell::Cell::new(false), |
| @@ -449,6 +462,8 @@ impl Lowerer { | |||
| 449 | self.blank(); | 462 | self.blank(); |
| 450 | } | 463 | } |
| 451 | 464 | ||
| 465 | + self.body_start = self.out.len(); | ||
| 466 | + | ||
| 452 | // Pass 4: bodies. | 467 | // Pass 4: bodies. |
| 453 | for (m, f) in files { | 468 | for (m, f) in files { |
| 454 | self.cur_mod = m.clone(); | 469 | self.cur_mod = m.clone(); |
| @@ -498,7 +513,13 @@ impl Lowerer { | |||
| 498 | // is needed can be discovered while lowering rather than predicted | 513 | // is needed can be discovered while lowering rather than predicted |
| 499 | // before it. The async dispatcher and the locking runtime each cost | 514 | // before it. The async dispatcher and the locking runtime each cost |
| 500 | // compile time, and most programs need neither. | 515 | // compile time, and most programs need neither. |
| 501 | - let body = std::mem::take(&mut self.out); | 516 | + // Hoisted items go between the declarations and the bodies: they are |
| 517 | + // top-level procs discovered while lowering a function. | ||
| 518 | + let mut body = std::mem::take(&mut self.out); | ||
| 519 | + if !self.hoisted.is_empty() { | ||
| 520 | + let at = self.body_start.min(body.len()); | ||
| 521 | + body.insert_str(at, &std::mem::take(&mut self.hoisted)); | ||
| 522 | + } | ||
| 502 | let mut out = String::from(include_str!("prelude.nim")); | 523 | let mut out = String::from(include_str!("prelude.nim")); |
| 503 | out.push('\n'); | 524 | out.push('\n'); |
| 504 | if self.uses_async { | 525 | if self.uses_async { |
| @@ -3122,6 +3143,109 @@ impl Lowerer { | |||
| 3122 | Ok(()) | 3143 | Ok(()) |
| 3123 | } | 3144 | } |
| 3124 | 3145 | ||
| 3146 | + /// Lower `thread::spawn(move || body)`. | ||
| 3147 | + /// | ||
| 3148 | + /// The body becomes a top-level `{.thread.}` proc and its captures become | ||
| 3149 | + /// a heap object passed as the argument -- which also carries the result, | ||
| 3150 | + /// since a Nim thread proc returns nothing. | ||
| 3151 | + fn spawn_thread(&mut self, cl: &syn::ExprClosure) -> Result<Val, String> { | ||
| 3152 | + if cl.capture.is_none() { | ||
| 3153 | + return Err("`thread::spawn` needs a `move` closure: the thread \ | ||
| 3154 | + outlives this scope, so its captures must be owned" | ||
| 3155 | + .into()); | ||
| 3156 | + } | ||
| 3157 | + if !cl.inputs.is_empty() { | ||
| 3158 | + return Err("`thread::spawn` takes a closure with no arguments".into()); | ||
| 3159 | + } | ||
| 3160 | + self.uses_sync.set(true); | ||
| 3161 | + self.uses_threads = true; | ||
| 3162 | + | ||
| 3163 | + // Whatever the body names that is bound out here is a capture. | ||
| 3164 | + let mut names = Vec::new(); | ||
| 3165 | + collect_idents(&cl.body, &mut names); | ||
| 3166 | + let mut caps: Vec<(String, Nim)> = Vec::new(); | ||
| 3167 | + for n in names { | ||
| 3168 | + if caps.iter().any(|(m, _)| *m == n) { | ||
| 3169 | + continue; | ||
| 3170 | + } | ||
| 3171 | + if let Some(t) = self.lookup(&n) { | ||
| 3172 | + caps.push((n, t)); | ||
| 3173 | + } else if let Some(Alias::Value { ty: Some(t), .. }) = self.lookup_alias(&n) { | ||
| 3174 | + caps.push((n, t)); | ||
| 3175 | + } | ||
| 3176 | + } | ||
| 3177 | + | ||
| 3178 | + let id = self.fresh("Spawn"); | ||
| 3179 | + let state = format!("{}State", id); | ||
| 3180 | + let body = match &*cl.body { | ||
| 3181 | + Expr::Block(b) => b.block.clone(), | ||
| 3182 | + other => syn::Block { | ||
| 3183 | + brace_token: Default::default(), | ||
| 3184 | + stmts: vec![Stmt::Expr(other.clone(), None)], | ||
| 3185 | + }, | ||
| 3186 | + }; | ||
| 3187 | + | ||
| 3188 | + // Lower the body against the capture object, into the hoisted buffer. | ||
| 3189 | + let outer_out = std::mem::take(&mut self.out); | ||
| 3190 | + let outer_indent = std::mem::replace(&mut self.indent, 0); | ||
| 3191 | + self.push_scope(); | ||
| 3192 | + for (n, t) in &caps { | ||
| 3193 | + self.bind_alias( | ||
| 3194 | + n, | ||
| 3195 | + Alias::Value { code: format!("a.{}", ident(n)), ty: Some(t.clone()) }, | ||
| 3196 | + ); | ||
| 3197 | + } | ||
| 3198 | + // The thread takes an opaque pointer so that every handle is the same | ||
| 3199 | + // Nim type; the capture object is recovered inside. | ||
| 3200 | + self.line(&format!("proc rsThread{}(p: pointer) {{.thread.}} =", id)); | ||
| 3201 | + self.indent += 1; | ||
| 3202 | + self.line(&format!("let a = cast[{}](p)", state)); | ||
| 3203 | + let outer_t = self.target.replace(("a.res[]".to_string(), None)); | ||
| 3204 | + let before = self.out.len(); | ||
| 3205 | + let tail = self.block_body(&body)?; | ||
| 3206 | + let out_ty = tail.as_ref().and_then(|v| v.ty.clone()); | ||
| 3207 | + self.emit_tail(tail); | ||
| 3208 | + if self.out.len() == before { | ||
| 3209 | + self.line("discard"); | ||
| 3210 | + } | ||
| 3211 | + self.target = outer_t; | ||
| 3212 | + self.pop_scope(); | ||
| 3213 | + self.indent = outer_indent; | ||
| 3214 | + let proc_text = std::mem::replace(&mut self.out, outer_out); | ||
| 3215 | + | ||
| 3216 | + // The state type has to be declared before the proc that takes it. | ||
| 3217 | + let res_ty = out_ty.clone().unwrap_or(Nim::Unit); | ||
| 3218 | + let mut decl = format!("type {}* = ref object of RootObj\n", state); | ||
| 3219 | + for (n, t) in &caps { | ||
| 3220 | + decl.push_str(&format!(" {}*: {}\n", ident(n), t.render())); | ||
| 3221 | + } | ||
| 3222 | + decl.push_str(&format!(" res*: ref {}\n\n", res_ty.render())); | ||
| 3223 | + self.hoisted.push_str(&decl); | ||
| 3224 | + self.hoisted.push_str(&proc_text); | ||
| 3225 | + self.hoisted.push('\n'); | ||
| 3226 | + | ||
| 3227 | + // Build the state, then start the thread. | ||
| 3228 | + let st = self.fresh("St"); | ||
| 3229 | + let h = self.fresh("H"); | ||
| 3230 | + self.line(&format!("var {}: {}", st, state)); | ||
| 3231 | + self.line(&format!("new({})", st)); | ||
| 3232 | + for (n, _) in &caps { | ||
| 3233 | + let v = match self.lookup_alias(n) { | ||
| 3234 | + Some(Alias::Value { code, .. }) => code, | ||
| 3235 | + _ => ident(n), | ||
| 3236 | + }; | ||
| 3237 | + self.line(&format!("{}.{} = {}", st, ident(n), v)); | ||
| 3238 | + } | ||
| 3239 | + self.line(&format!("let {} = rsNewJoinHandle[{}]()", h, res_ty.render())); | ||
| 3240 | + self.line(&format!("{}.res = {}.res", st, h)); | ||
| 3241 | + self.line(&format!("{}.st = RootRef({})", h, st)); | ||
| 3242 | + self.line(&format!( | ||
| 3243 | + "createThread({}.th, rsThread{}, cast[pointer]({}))", | ||
| 3244 | + h, id, st | ||
| 3245 | + )); | ||
| 3246 | + Ok(Val::new(h, Some(Nim::Named("RsJoinHandle".into(), vec![res_ty])))) | ||
| 3247 | + } | ||
| 3248 | + | ||
| 3125 | /// Bind one element of an iterator under a fresh name, and return how to | 3249 | /// Bind one element of an iterator under a fresh name, and return how to |
| 3126 | /// refer to it and what type it has. | 3250 | /// refer to it and what type it has. |
| 3127 | fn bind_element(&mut self, it: &Iter, i: &str) -> Result<(String, Option<Nim>), String> { | 3251 | fn bind_element(&mut self, it: &Iter, i: &str) -> Result<(String, Option<Nim>), String> { |
| @@ -4723,6 +4847,16 @@ impl Lowerer { | |||
| 4723 | return Err("only calls to named functions are supported".into()); | 4847 | return Err("only calls to named functions are supported".into()); |
| 4724 | }; | 4848 | }; |
| 4725 | let name = path_name(&p.path); | 4849 | let name = path_name(&p.path); |
| 4850 | + | ||
| 4851 | + // `thread::spawn(move || ..)` is handled before the arguments are | ||
| 4852 | + // lowered: a `move` closure is rejected everywhere else, and here it | ||
| 4853 | + // is the whole point. See `src/prelude_sync.nim`. | ||
| 4854 | + if name == "spawn" && c.args.len() == 1 { | ||
| 4855 | + if let Expr::Closure(cl) = &c.args[0] { | ||
| 4856 | + return self.spawn_thread(cl); | ||
| 4857 | + } | ||
| 4858 | + } | ||
| 4859 | + | ||
| 4726 | let target = self.resolve_fn(&p.path); | 4860 | let target = self.resolve_fn(&p.path); |
| 4727 | let ptys: Vec<Nim> = target | 4861 | let ptys: Vec<Nim> = target |
| 4728 | .as_ref() | 4862 | .as_ref() |
| @@ -5227,6 +5361,11 @@ impl Lowerer { | |||
| 5227 | // which would look for a `Result`. | 5361 | // which would look for a `Result`. |
| 5228 | if matches!(name.as_str(), "unwrap" | "expect") { | 5362 | if matches!(name.as_str(), "unwrap" | "expect") { |
| 5229 | if let Some(Nim::Named(n, a)) = &rt { | 5363 | if let Some(Nim::Named(n, a)) = &rt { |
| 5364 | + // `join()` already produced the value; Rust wraps it in a | ||
| 5365 | + // `Result` only because a panicking thread is caught there. | ||
| 5366 | + if n == "RsJoinOut" { | ||
| 5367 | + return Ok(Val::new(recv.code.clone(), a.first().cloned())); | ||
| 5368 | + } | ||
| 5230 | if n == "RsGuard" { | 5369 | if n == "RsGuard" { |
| 5231 | let inner = a.first().cloned(); | 5370 | let inner = a.first().cloned(); |
| 5232 | let mut v = Val::new(recv.code.clone(), inner.clone()); | 5371 | let mut v = Val::new(recv.code.clone(), inner.clone()); |
| @@ -5458,6 +5597,18 @@ impl Lowerer { | |||
| 5458 | Some(Nim::Ptr(Box::new(elem))), | 5597 | Some(Nim::Ptr(Box::new(elem))), |
| 5459 | ) | 5598 | ) |
| 5460 | } | 5599 | } |
| 5600 | + // `h.join()` waits and yields what the thread produced. Rust's | ||
| 5601 | + // returns `Result` because a panicking thread is caught there; | ||
| 5602 | + // here a panic ends the process, so there is no error case -- see | ||
| 5603 | + // `src/prelude_sync.nim`. | ||
| 5604 | + "join" if matches!(&rt, Some(Nim::Named(n, _)) if n == "RsJoinHandle") => { | ||
| 5605 | + let Some(Nim::Named(_, a)) = &rt else { unreachable!() }; | ||
| 5606 | + let out = a.first().cloned(); | ||
| 5607 | + return Ok(Val::new( | ||
| 5608 | + format!("rsJoin({})", recv.code), | ||
| 5609 | + Some(Nim::Named("RsJoinOut".into(), out.into_iter().collect())), | ||
| 5610 | + )); | ||
| 5611 | + } | ||
| 5461 | // Atomic operations. Each is the same C11 primitive on both | 5612 | // Atomic operations. Each is the same C11 primitive on both |
| 5462 | // sides, with the ordering carried through unchanged. | 5613 | // sides, with the ordering carried through unchanged. |
| 5463 | "load" | "store" | "swap" | "fetch_add" | "fetch_sub" | "fetch_and" | 5614 | "load" | "store" | "swap" | "fetch_add" | "fetch_sub" | "fetch_and" |
| @@ -6108,6 +6259,22 @@ fn module_of(prefix: &[String]) -> String { | |||
| 6108 | } | 6259 | } |
| 6109 | } | 6260 | } |
| 6110 | 6261 | ||
| 6262 | +/// Every single-segment identifier an expression mentions. Used to find what | ||
| 6263 | +/// a `move` closure captures: a name it uses that is bound outside it. | ||
| 6264 | +fn collect_idents(e: &Expr, out: &mut Vec<String>) { | ||
| 6265 | + use syn::visit::Visit; | ||
| 6266 | + struct V<'a>(&'a mut Vec<String>); | ||
| 6267 | + impl<'ast> Visit<'ast> for V<'_> { | ||
| 6268 | + fn visit_expr_path(&mut self, p: &'ast syn::ExprPath) { | ||
| 6269 | + if p.path.segments.len() == 1 && p.qself.is_none() { | ||
| 6270 | + self.0.push(p.path.segments[0].ident.to_string()); | ||
| 6271 | + } | ||
| 6272 | + syn::visit::visit_expr_path(self, p); | ||
| 6273 | + } | ||
| 6274 | + } | ||
| 6275 | + V(out).visit_expr(e); | ||
| 6276 | +} | ||
| 6277 | + | ||
| 6111 | /// Whether a type involves the locking runtime. | 6278 | /// Whether a type involves the locking runtime. |
| 6112 | fn mentions_sync(t: &Nim) -> bool { | 6279 | fn mentions_sync(t: &Nim) -> bool { |
| 6113 | match t { | 6280 | match t { |
modified
src/main.rs +29 -2 | @@ -11,6 +11,16 @@ mod ty; | ||
| 11 | 11 | use std::path::PathBuf; |
| 12 | 12 | use std::process::ExitCode; |
| 13 | 13 | |
| 14 | +/// Emitted beside a program that spawns threads. Nim reads `<file>.nim.cfg` | |
| 15 | +/// automatically, so the build requirement travels with the output rather | |
| 16 | +/// than being a line in a README that someone has to read. | |
| 17 | +const THREAD_CFG: &str = "\ | |
| 18 | +# This program shares reference-counted values across threads, which needs | |
| 19 | +# atomic reference counting. Nim's default is not atomic, and the failure is a | |
| 20 | +# segfault in the cycle collector rather than anything that names the cause. | |
| 21 | +--mm:atomicArc | |
| 22 | +"; | |
| 23 | + | |
| 14 | 24 | fn main() -> ExitCode { |
| 15 | 25 | match run() { |
| 16 | 26 | Ok(()) => ExitCode::SUCCESS, |
| @@ -99,8 +109,25 @@ fn run() -> Result<(), String> { | ||
| 99 | 109 | let nim = lowerer.lower_file(&files)?; |
| 100 | 110 | |
| 101 | 111 | match output { |
| 102 | - Some(p) => std::fs::write(&p, nim) | |
| 103 | - .map_err(|e| format!("cannot write {}: {e}", p.display()))?, | |
| 112 | + Some(p) => { | |
| 113 | + std::fs::write(&p, nim) | |
| 114 | + .map_err(|e| format!("cannot write {}: {e}", p.display()))?; | |
| 115 | + // Sharing a `ref` across threads needs atomic reference counting. | |
| 116 | + // Nim's default is not atomic -- which segfaults in the cycle | |
| 117 | + // collector rather than failing quietly -- and the setting is a | |
| 118 | + // compiler flag, not anything the source can say. Nim reads | |
| 119 | + // `<file>.nim.cfg` automatically, so the requirement travels with | |
| 120 | + // the output instead of being a line in a README. | |
| 121 | + let cfg = p.with_extension("nim.cfg"); | |
| 122 | + if lowerer.uses_threads { | |
| 123 | + std::fs::write(&cfg, THREAD_CFG) | |
| 124 | + .map_err(|e| format!("cannot write {}: {e}", cfg.display()))?; | |
| 125 | + } else { | |
| 126 | + // A stale config from an earlier run would silently change how | |
| 127 | + // this one is built. | |
| 128 | + let _ = std::fs::remove_file(&cfg); | |
| 129 | + } | |
| 130 | + } | |
| 104 | 131 | None => print!("{nim}"), |
| 105 | 132 | } |
| 106 | 133 | Ok(()) |
| @@ -11,6 +11,16 @@ mod ty; | |||
| 11 | use std::path::PathBuf; | 11 | use std::path::PathBuf; |
| 12 | use std::process::ExitCode; | 12 | use std::process::ExitCode; |
| 13 | 13 | ||
| 14 | +/// Emitted beside a program that spawns threads. Nim reads `<file>.nim.cfg` | ||
| 15 | +/// automatically, so the build requirement travels with the output rather | ||
| 16 | +/// than being a line in a README that someone has to read. | ||
| 17 | +const THREAD_CFG: &str = "\ | ||
| 18 | +# This program shares reference-counted values across threads, which needs | ||
| 19 | +# atomic reference counting. Nim's default is not atomic, and the failure is a | ||
| 20 | +# segfault in the cycle collector rather than anything that names the cause. | ||
| 21 | +--mm:atomicArc | ||
| 22 | +"; | ||
| 23 | + | ||
| 14 | fn main() -> ExitCode { | 24 | fn main() -> ExitCode { |
| 15 | match run() { | 25 | match run() { |
| 16 | Ok(()) => ExitCode::SUCCESS, | 26 | Ok(()) => ExitCode::SUCCESS, |
| @@ -99,8 +109,25 @@ fn run() -> Result<(), String> { | |||
| 99 | let nim = lowerer.lower_file(&files)?; | 109 | let nim = lowerer.lower_file(&files)?; |
| 100 | 110 | ||
| 101 | match output { | 111 | match output { |
| 102 | - Some(p) => std::fs::write(&p, nim) | 112 | + Some(p) => { |
| 103 | - .map_err(|e| format!("cannot write {}: {e}", p.display()))?, | 113 | + std::fs::write(&p, nim) |
| 114 | + .map_err(|e| format!("cannot write {}: {e}", p.display()))?; | ||
| 115 | + // Sharing a `ref` across threads needs atomic reference counting. | ||
| 116 | + // Nim's default is not atomic -- which segfaults in the cycle | ||
| 117 | + // collector rather than failing quietly -- and the setting is a | ||
| 118 | + // compiler flag, not anything the source can say. Nim reads | ||
| 119 | + // `<file>.nim.cfg` automatically, so the requirement travels with | ||
| 120 | + // the output instead of being a line in a README. | ||
| 121 | + let cfg = p.with_extension("nim.cfg"); | ||
| 122 | + if lowerer.uses_threads { | ||
| 123 | + std::fs::write(&cfg, THREAD_CFG) | ||
| 124 | + .map_err(|e| format!("cannot write {}: {e}", cfg.display()))?; | ||
| 125 | + } else { | ||
| 126 | + // A stale config from an earlier run would silently change how | ||
| 127 | + // this one is built. | ||
| 128 | + let _ = std::fs::remove_file(&cfg); | ||
| 129 | + } | ||
| 130 | + } | ||
| 104 | None => print!("{nim}"), | 131 | None => print!("{nim}"), |
| 105 | } | 132 | } |
| 106 | Ok(()) | 133 | Ok(()) |
modified
src/prelude_sync.nim +37 -0 | @@ -60,3 +60,40 @@ proc rsMutexNew*[T](v: T): RsMutex[T] = | ||
| 60 | 60 | new(result) |
| 61 | 61 | initLock(result.lk) |
| 62 | 62 | result.val = v |
| 63 | + | |
| 64 | + | |
| 65 | +# --------------------------------------------------------------------------- | |
| 66 | +# Threads. | |
| 67 | +# | |
| 68 | +# Rust's `thread::spawn` takes a closure that captures. A Nim thread proc is | |
| 69 | +# `{.thread, nimcall.}` and deliberately cannot -- a closure refers to the | |
| 70 | +# creating thread's environment. So each spawn site is closure-converted: the | |
| 71 | +# body becomes a top-level `{.thread.}` proc, and whatever it captured becomes | |
| 72 | +# a heap object passed as the argument, which is also where the result is put | |
| 73 | +# because a Nim thread proc returns nothing. | |
| 74 | +# | |
| 75 | +# `move` is what Rust requires for `spawn`, and copying the captures into that | |
| 76 | +# object is exactly what `move` means. | |
| 77 | +# | |
| 78 | +# What differs: **a panicking thread is not isolated.** Rust's `join` returns | |
| 79 | +# `Err` when the thread panicked and the process continues. Panics here map to | |
| 80 | +# `quit(101)`, which ends everything. A program that relies on catching a | |
| 81 | +# thread panic would differ. | |
| 82 | +# --------------------------------------------------------------------------- | |
| 83 | + | |
| 84 | +# The handle is uniform in `T` so that `Vec<JoinHandle<T>>` is one Nim type: | |
| 85 | +# the per-site capture object is passed to the thread as an opaque pointer and | |
| 86 | +# kept alive by `st`, while the result lands in a slot both sides share. | |
| 87 | + | |
| 88 | +type RsJoinHandle*[T] = ref object | |
| 89 | + th*: Thread[pointer] | |
| 90 | + st*: RootRef ## keeps the capture object alive while the thread runs | |
| 91 | + res*: ref T | |
| 92 | + | |
| 93 | +proc rsNewJoinHandle*[T](): RsJoinHandle[T] = | |
| 94 | + new(result) | |
| 95 | + new(result.res) | |
| 96 | + | |
| 97 | +proc rsJoin*[T](h: RsJoinHandle[T]): T = | |
| 98 | + joinThread(h.th) | |
| 99 | + h.res[] | |
| @@ -60,3 +60,40 @@ proc rsMutexNew*[T](v: T): RsMutex[T] = | |||
| 60 | new(result) | 60 | new(result) |
| 61 | initLock(result.lk) | 61 | initLock(result.lk) |
| 62 | result.val = v | 62 | result.val = v |
| 63 | + | ||
| 64 | + | ||
| 65 | +# --------------------------------------------------------------------------- | ||
| 66 | +# Threads. | ||
| 67 | +# | ||
| 68 | +# Rust's `thread::spawn` takes a closure that captures. A Nim thread proc is | ||
| 69 | +# `{.thread, nimcall.}` and deliberately cannot -- a closure refers to the | ||
| 70 | +# creating thread's environment. So each spawn site is closure-converted: the | ||
| 71 | +# body becomes a top-level `{.thread.}` proc, and whatever it captured becomes | ||
| 72 | +# a heap object passed as the argument, which is also where the result is put | ||
| 73 | +# because a Nim thread proc returns nothing. | ||
| 74 | +# | ||
| 75 | +# `move` is what Rust requires for `spawn`, and copying the captures into that | ||
| 76 | +# object is exactly what `move` means. | ||
| 77 | +# | ||
| 78 | +# What differs: **a panicking thread is not isolated.** Rust's `join` returns | ||
| 79 | +# `Err` when the thread panicked and the process continues. Panics here map to | ||
| 80 | +# `quit(101)`, which ends everything. A program that relies on catching a | ||
| 81 | +# thread panic would differ. | ||
| 82 | +# --------------------------------------------------------------------------- | ||
| 83 | + | ||
| 84 | +# The handle is uniform in `T` so that `Vec<JoinHandle<T>>` is one Nim type: | ||
| 85 | +# the per-site capture object is passed to the thread as an opaque pointer and | ||
| 86 | +# kept alive by `st`, while the result lands in a slot both sides share. | ||
| 87 | + | ||
| 88 | +type RsJoinHandle*[T] = ref object | ||
| 89 | + th*: Thread[pointer] | ||
| 90 | + st*: RootRef ## keeps the capture object alive while the thread runs | ||
| 91 | + res*: ref T | ||
| 92 | + | ||
| 93 | +proc rsNewJoinHandle*[T](): RsJoinHandle[T] = | ||
| 94 | + new(result) | ||
| 95 | + new(result.res) | ||
| 96 | + | ||
| 97 | +proc rsJoin*[T](h: RsJoinHandle[T]): T = | ||
| 98 | + joinThread(h.th) | ||
| 99 | + h.res[] | ||
modified
src/ty.rs +1 -0 | @@ -292,6 +292,7 @@ pub fn map(t: &Type) -> Result<Nim, String> { | ||
| 292 | 292 | // Shared ownership and locking: see `src/prelude_sync.nim`. |
| 293 | 293 | ("Arc", 1) | ("Rc", 1) => Ok(Nim::Named("RsArc".into(), args)), |
| 294 | 294 | ("Mutex", 1) => Ok(Nim::Named("RsMutex".into(), args)), |
| 295 | + ("JoinHandle", 1) => Ok(Nim::Named("RsJoinHandle".into(), args)), | |
| 295 | 296 | ("Vec", 1) => Ok(Nim::Seq(Box::new(args[0].clone().owned()))), |
| 296 | 297 | ("Option", 1) => Ok(Nim::Named("Option".into(), args)), |
| 297 | 298 | ("Result", 2) => Ok(Nim::Named("Result".into(), args)), |
| @@ -292,6 +292,7 @@ pub fn map(t: &Type) -> Result<Nim, String> { | |||
| 292 | // Shared ownership and locking: see `src/prelude_sync.nim`. | 292 | // Shared ownership and locking: see `src/prelude_sync.nim`. |
| 293 | ("Arc", 1) | ("Rc", 1) => Ok(Nim::Named("RsArc".into(), args)), | 293 | ("Arc", 1) | ("Rc", 1) => Ok(Nim::Named("RsArc".into(), args)), |
| 294 | ("Mutex", 1) => Ok(Nim::Named("RsMutex".into(), args)), | 294 | ("Mutex", 1) => Ok(Nim::Named("RsMutex".into(), args)), |
| 295 | + ("JoinHandle", 1) => Ok(Nim::Named("RsJoinHandle".into(), args)), | ||
| 295 | ("Vec", 1) => Ok(Nim::Seq(Box::new(args[0].clone().owned()))), | 296 | ("Vec", 1) => Ok(Nim::Seq(Box::new(args[0].clone().owned()))), |
| 296 | ("Option", 1) => Ok(Nim::Named("Option".into(), args)), | 297 | ("Option", 1) => Ok(Nim::Named("Option".into(), args)), |
| 297 | ("Result", 2) => Ok(Nim::Named("Result".into(), args)), | 298 | ("Result", 2) => Ok(Nim::Named("Result".into(), args)), |
added
tests/cases/046-threads.rs +79 -0 | new file mode 100644 | ||
| @@ -0,0 +1,79 @@ | ||
| 1 | +// Rust's `thread::spawn` takes a closure that captures. A Nim thread proc is | |
| 2 | +// `{.thread, nimcall.}` and deliberately cannot, since a closure refers to the | |
| 3 | +// creating thread's environment. So each spawn site is closure-converted: the | |
| 4 | +// body becomes a top-level `{.thread.}` proc and the captures become a heap | |
| 5 | +// object passed as its argument -- which also carries the result, because a | |
| 6 | +// Nim thread proc returns nothing. | |
| 7 | +// | |
| 8 | +// `move` is what Rust requires here, and copying the captures into that object | |
| 9 | +// is exactly what `move` means. | |
| 10 | +// | |
| 11 | +// The totals below are deterministic despite the concurrency, which is what | |
| 12 | +// makes this a differential test rather than a coin flip. | |
| 13 | + | |
| 14 | +use std::sync::atomic::{AtomicUsize, Ordering}; | |
| 15 | +use std::sync::{Arc, Mutex}; | |
| 16 | +use std::thread; | |
| 17 | + | |
| 18 | +fn main() { | |
| 19 | + // A value computed on another thread and handed back by join. | |
| 20 | + let h = thread::spawn(move || { | |
| 21 | + let mut t: i32 = 0; | |
| 22 | + let mut i: i32 = 0; | |
| 23 | + while i < 100 { | |
| 24 | + t += i; | |
| 25 | + i += 1; | |
| 26 | + } | |
| 27 | + t | |
| 28 | + }); | |
| 29 | + println!("{}", h.join().unwrap()); | |
| 30 | + | |
| 31 | + // A captured value, moved in. | |
| 32 | + let base: i32 = 1000; | |
| 33 | + let h2 = thread::spawn(move || base * 2); | |
| 34 | + println!("{}", h2.join().unwrap()); | |
| 35 | + | |
| 36 | + // Shared mutable state behind Arc<Mutex<_>>: the total is fixed even | |
| 37 | + // though the interleaving is not. | |
| 38 | + let counter: Arc<Mutex<i32>> = Arc::new(Mutex::new(0)); | |
| 39 | + let mut handles: Vec<thread::JoinHandle<i32>> = Vec::new(); | |
| 40 | + let mut k = 0; | |
| 41 | + while k < 4 { | |
| 42 | + let c = counter.clone(); | |
| 43 | + handles.push(thread::spawn(move || { | |
| 44 | + let mut n = 0; | |
| 45 | + while n < 1000 { | |
| 46 | + let mut g = c.lock().unwrap(); | |
| 47 | + *g += 1; | |
| 48 | + n += 1; | |
| 49 | + } | |
| 50 | + 0 | |
| 51 | + })); | |
| 52 | + k += 1; | |
| 53 | + } | |
| 54 | + for h in handles { | |
| 55 | + let _ = h.join().unwrap(); | |
| 56 | + } | |
| 57 | + println!("{}", *counter.lock().unwrap()); | |
| 58 | + | |
| 59 | + // The same, through an atomic. | |
| 60 | + let hits: Arc<AtomicUsize> = Arc::new(AtomicUsize::new(0)); | |
| 61 | + let mut hs2: Vec<thread::JoinHandle<i32>> = Vec::new(); | |
| 62 | + let mut j = 0; | |
| 63 | + while j < 4 { | |
| 64 | + let a = hits.clone(); | |
| 65 | + hs2.push(thread::spawn(move || { | |
| 66 | + let mut n = 0; | |
| 67 | + while n < 1000 { | |
| 68 | + a.fetch_add(1, Ordering::Relaxed); | |
| 69 | + n += 1; | |
| 70 | + } | |
| 71 | + 0 | |
| 72 | + })); | |
| 73 | + j += 1; | |
| 74 | + } | |
| 75 | + for h in hs2 { | |
| 76 | + let _ = h.join().unwrap(); | |
| 77 | + } | |
| 78 | + println!("{}", hits.load(Ordering::SeqCst)); | |
| 79 | +} | |
| new file mode 100644 | |||
| @@ -0,0 +1,79 @@ | |||
| 1 | +// Rust's `thread::spawn` takes a closure that captures. A Nim thread proc is | ||
| 2 | +// `{.thread, nimcall.}` and deliberately cannot, since a closure refers to the | ||
| 3 | +// creating thread's environment. So each spawn site is closure-converted: the | ||
| 4 | +// body becomes a top-level `{.thread.}` proc and the captures become a heap | ||
| 5 | +// object passed as its argument -- which also carries the result, because a | ||
| 6 | +// Nim thread proc returns nothing. | ||
| 7 | +// | ||
| 8 | +// `move` is what Rust requires here, and copying the captures into that object | ||
| 9 | +// is exactly what `move` means. | ||
| 10 | +// | ||
| 11 | +// The totals below are deterministic despite the concurrency, which is what | ||
| 12 | +// makes this a differential test rather than a coin flip. | ||
| 13 | + | ||
| 14 | +use std::sync::atomic::{AtomicUsize, Ordering}; | ||
| 15 | +use std::sync::{Arc, Mutex}; | ||
| 16 | +use std::thread; | ||
| 17 | + | ||
| 18 | +fn main() { | ||
| 19 | + // A value computed on another thread and handed back by join. | ||
| 20 | + let h = thread::spawn(move || { | ||
| 21 | + let mut t: i32 = 0; | ||
| 22 | + let mut i: i32 = 0; | ||
| 23 | + while i < 100 { | ||
| 24 | + t += i; | ||
| 25 | + i += 1; | ||
| 26 | + } | ||
| 27 | + t | ||
| 28 | + }); | ||
| 29 | + println!("{}", h.join().unwrap()); | ||
| 30 | + | ||
| 31 | + // A captured value, moved in. | ||
| 32 | + let base: i32 = 1000; | ||
| 33 | + let h2 = thread::spawn(move || base * 2); | ||
| 34 | + println!("{}", h2.join().unwrap()); | ||
| 35 | + | ||
| 36 | + // Shared mutable state behind Arc<Mutex<_>>: the total is fixed even | ||
| 37 | + // though the interleaving is not. | ||
| 38 | + let counter: Arc<Mutex<i32>> = Arc::new(Mutex::new(0)); | ||
| 39 | + let mut handles: Vec<thread::JoinHandle<i32>> = Vec::new(); | ||
| 40 | + let mut k = 0; | ||
| 41 | + while k < 4 { | ||
| 42 | + let c = counter.clone(); | ||
| 43 | + handles.push(thread::spawn(move || { | ||
| 44 | + let mut n = 0; | ||
| 45 | + while n < 1000 { | ||
| 46 | + let mut g = c.lock().unwrap(); | ||
| 47 | + *g += 1; | ||
| 48 | + n += 1; | ||
| 49 | + } | ||
| 50 | + 0 | ||
| 51 | + })); | ||
| 52 | + k += 1; | ||
| 53 | + } | ||
| 54 | + for h in handles { | ||
| 55 | + let _ = h.join().unwrap(); | ||
| 56 | + } | ||
| 57 | + println!("{}", *counter.lock().unwrap()); | ||
| 58 | + | ||
| 59 | + // The same, through an atomic. | ||
| 60 | + let hits: Arc<AtomicUsize> = Arc::new(AtomicUsize::new(0)); | ||
| 61 | + let mut hs2: Vec<thread::JoinHandle<i32>> = Vec::new(); | ||
| 62 | + let mut j = 0; | ||
| 63 | + while j < 4 { | ||
| 64 | + let a = hits.clone(); | ||
| 65 | + hs2.push(thread::spawn(move || { | ||
| 66 | + let mut n = 0; | ||
| 67 | + while n < 1000 { | ||
| 68 | + a.fetch_add(1, Ordering::Relaxed); | ||
| 69 | + n += 1; | ||
| 70 | + } | ||
| 71 | + 0 | ||
| 72 | + })); | ||
| 73 | + j += 1; | ||
| 74 | + } | ||
| 75 | + for h in hs2 { | ||
| 76 | + let _ = h.join().unwrap(); | ||
| 77 | + } | ||
| 78 | + println!("{}", hits.load(Ordering::SeqCst)); | ||
| 79 | +} | ||