Source file `concurrent_in_thread.ml`

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
open Base
open Await
module Expert = Multicore

let rec spawn : type (r : value_or_null) a. (r, a, unit) Concurrent.Scheduler.spawn_fn =
  fun scope #{ fn; name; affinity } resource ->
  let spawn =
    match affinity with
    | Null -> Expert.spawn
    | This domain -> Expert.spawn_on ~domain:(domain % Expert.max_domains ())
  in
  match
    spawn
      (fun ({ many = token }, r) ->
        (* Always record backtraces in concurrent threads *)
        Stdlib.Printexc.record_backtrace true;
        Or_null.iter name ~f:Thread.set_current_thread_name;
        Concurrent.Scope.Token.use token ~f:(fun terminator scope ->
          with_blocking terminator ~f:(fun [@inline] c -> fn scope () c r [@nontail])
          [@nontail])
        [@nontail])
      ({ many = Concurrent.Scope.add scope }, resource)
  with
  | Spawned -> Spawned
  | Failed (({ many = token }, r), exn, bt) ->
    Concurrent.Scope.Token.drop token;
    Failed (r, exn, bt)

and[@inline] create await = exclave_
  (Concurrent.create [@mode portable])
    await
    ~scheduler:((Concurrent.Scheduler.create [@mode portable]) ~spawn)

and[@inline] with_blocking
  : 'r.
  Terminator.t @ local -> f:(unit Concurrent.t @ local portable -> 'r) @ local once -> 'r
  =
  fun terminator ~f ->
  f
    (create
       (Await.Expert.create
          ~sync:Sync.blocking
          ~terminator:(Terminator.Expert.globalize terminator))) [@nontail]
;;

let scheduler = (Concurrent.Scheduler.create [@mode portable]) ~spawn

let[@inline] in_scope await scope = exclave_
  (Concurrent.Spawn.create [@mode portable]) ~scope (create await)
;;

let[@inline] spawn_into spawn = exclave_
  in_scope (Concurrent.Spawn.await spawn) (Concurrent.Spawn.scope spawn)
;;