terminal user interface to jujutsu. Focused on speed and clarity
update to use latest concurrent lwd fork
+1226
-923
-19
forks/lwd/.vscode/launch.json
-19
forks/lwd/.vscode/launch.json
···
1
-
{
2
-
// Use IntelliSense to learn about possible attributes.
3
-
// Hover to view descriptions of existing attributes.
4
-
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
5
-
"version": "0.2.0",
6
-
"configurations": [
7
-
{
8
-
"name": "test_program",
9
-
"type": "ocaml.earlybird",
10
-
"request": "launch",
11
-
"stopOnEntry": true,
12
-
"console": "integratedTerminal",
13
-
"program": "${workspaceFolder}/_build/default/examples/interact/test_program.bc",
14
-
"onlyDebugGlob": "<${workspaceFolder}/**/*>",
15
-
"yieldSteps": 1024,
16
-
"cwd": "${workspaceFolder}"
17
-
}
18
-
]
19
-
}
-7
forks/lwd/.vscode/settings.json
-7
forks/lwd/.vscode/settings.json
forks/lwd/dune
forks/lwd/dune
This is a binary file and will not be displayed.
+9
-1
forks/lwd/dune-project
+9
-1
forks/lwd/dune-project
···
20
20
(name lwd)
21
21
(synopsis "Lightweight reactive documents")
22
22
(documentation "https://let-def.github.io/lwd/doc")
23
+
23
24
(depends
24
25
dune
25
26
seq
···
35
36
(depends
36
37
lwd
37
38
(picos (>= "0.6.0"))
38
-
(picos_std (>= "0.6.0"))))
39
+
(picos_std (>= "0.6.0"))))
40
+
(package
41
+
(name lwd_stdlib)
42
+
(synopsis "Lightweight reactive documents with Stdlib backend")
43
+
(allow_empty)
44
+
(depends
45
+
lwd
46
+
))
-67
forks/lwd/lib/lwd/custom_mutex.ml
-67
forks/lwd/lib/lwd/custom_mutex.ml
···
1
-
module type MUTEX = sig
2
-
(** Locks for mutual exclusion with support for multiple concurrency backends.
3
-
4
-
This module provides a unified interface for mutexes that can work with
5
-
different concurrency backends (OCaml standard library, Picos, etc.).
6
-
It extends the standard OCaml mutex interface with additional functionality
7
-
for acquiring multiple mutexes atomically.
8
-
*)
9
-
10
-
type t
11
-
(** The type of mutexes. *)
12
-
13
-
val create : unit -> t
14
-
(** Return a new mutex. *)
15
-
16
-
val lock : t -> unit
17
-
(** Lock the given mutex. Only one thread can have the mutex locked
18
-
at any time. A thread that attempts to lock a mutex already locked
19
-
by another thread will suspend until the other thread unlocks
20
-
the mutex.
21
-
22
-
@raise Sys_error if the mutex is already locked by the thread calling
23
-
{!lock}.
24
-
25
-
@before 4.12 {!Sys_error} was not raised for recursive locking
26
-
(platform-dependent behaviour) *)
27
-
28
-
val try_lock : t -> bool
29
-
(** Same as {!lock}, but does not suspend the calling thread if
30
-
the mutex is already locked: just return [false] immediately
31
-
in that case. If the mutex is unlocked, lock it and
32
-
return [true]. *)
33
-
34
-
val unlock : t -> unit
35
-
(** Unlock the given mutex. Other threads suspended trying to lock
36
-
the mutex will restart. The mutex must have been previously locked
37
-
by the thread that calls {!unlock}.
38
-
@raise Sys_error if the mutex is unlocked or was locked by another thread.
39
-
40
-
@before 4.12 {!Sys_error} was not raised when unlocking an unlocked mutex
41
-
or when unlocking a mutex from a different thread. *)
42
-
43
-
val protect : t -> (unit -> 'a) -> 'a
44
-
(** [protect mutex f] runs [f()] in a critical section where [mutex]
45
-
is locked (using {!lock}); it then takes care of releasing [mutex],
46
-
whether [f()] returned a value or raised an exception.
47
-
48
-
The unlocking operation is guaranteed to always takes place,
49
-
even in the event an asynchronous exception (e.g. {!Sys.Break}) is raised
50
-
in some signal handler.
51
-
52
-
@since 5.1 *)
53
-
54
-
val lock_all : t list -> bool
55
-
(** [lock_all mutexes] attempts to acquire all mutexes in the list atomically.
56
-
It uses {!try_lock} for each mutex in the order provided. If any mutex
57
-
cannot be acquired, it releases all previously acquired mutexes and
58
-
returns [false]. If all mutexes are successfully acquired, it returns [true].
59
-
60
-
This function is useful for avoiding deadlocks when multiple mutexes
61
-
need to be acquired simultaneously.
62
-
63
-
@return [true] if all mutexes were successfully acquired, [false] otherwise.
64
-
65
-
Note: The caller is responsible for unlocking all mutexes that were
66
-
successfully acquired when this function returns [true]. *)
67
-
end
forks/lwd/lib/lwd/custom_mutex.mli
forks/lwd/lib/lwd/custom_mutex.mli
This is a binary file and will not be displayed.
+4
-2
forks/lwd/lib/lwd/dune
+4
-2
forks/lwd/lib/lwd/dune
···
1
1
(library
2
2
(name lwd)
3
3
(public_name lwd)
4
-
(modules lwd lwd_seq lwd_table lwd_infix lwd_utils mutex_stdlib lwd_impl mutex_backend )
5
-
(libraries seq)
4
+
(modules lwd lwd_impl lwd_seq lwd_table lwd_infix lwd_utils mutex_backend )
5
+
(virtual_modules lwd)
6
+
; (default_implementation lwd_stdlib)
7
+
(libraries seq logs)
6
8
(inline_tests
7
9
(backend qtest.lib)
8
10
(executable
-2
forks/lwd/lib/lwd/lwd.ml
-2
forks/lwd/lib/lwd/lwd.ml
+2
forks/lwd/lib/lwd/lwd.mli
+2
forks/lwd/lib/lwd/lwd.mli
+1101
-753
forks/lwd/lib/lwd/lwd_impl.ml
+1101
-753
forks/lwd/lib/lwd/lwd_impl.ml
···
1
1
module Make (Mutex : Mutex_backend.MUTEX) = struct
2
+
let log_src = Logs.Src.create "lwd.impl" ~doc:"Lwd implementation"
3
+
module Log = (val Logs.src_log log_src : Logs.LOG)
2
4
3
-
(** Create-only version of [Obj.t] *)
4
-
module Any : sig
5
-
type t
6
-
val any : 'a -> t
7
-
end = struct
8
-
type t = Obj.t
9
-
let any = Obj.repr
10
-
end
5
+
(** Create-only version of [Obj.t] *)
6
+
module Any : sig
7
+
type t
8
+
val any : 'a -> t
9
+
end = struct
10
+
type t = Obj.t
11
+
let any = Obj.repr
12
+
end
13
+
14
+
type 'a eval =
15
+
| Eval_none
16
+
| Eval_progress
17
+
| Eval_some of 'a
18
+
| Eval_invalid_next
19
+
20
+
type 'a t_ =
21
+
| Pure of 'a
22
+
| Operator : {
23
+
mutex : Mutex.t;
24
+
mutable value : 'a eval; (* cached value *)
25
+
mutable trace : trace; (* list of parents this can invalidate *)
26
+
mutable trace_idx : trace_idx; (* list of direct children that can invalidate this *)
27
+
desc: 'a desc;
28
+
} -> 'a t_
29
+
| Root : {
30
+
mutex : Mutex.t;
31
+
mutable value : 'a eval; (* cached value *)
32
+
mutable trace_idx : trace_idx; (* list of direct children that can invalidate this *)
33
+
mutable on_invalidate : 'a -> unit;
34
+
mutable acquired : bool;
35
+
child : 'a t_;
36
+
} -> 'a t_
37
+
38
+
and _ desc =
39
+
| Map : 'a t_ * ('a -> 'b) -> 'b desc
40
+
| Map2 : 'a t_ * 'b t_ * ('a -> 'b -> 'c) -> 'c desc
41
+
| Pair : 'a t_ * 'b t_ -> ('a * 'b) desc
42
+
| App : ('a -> 'b) t_ * 'a t_ -> 'b desc
43
+
| Join : { child : 'a t_ t_; mutable intermediate : 'a t_ option } -> 'a desc
44
+
| Var : { mutable binding : 'a } -> 'a desc
45
+
| Prim : { acquire : 'a t -> 'a;
46
+
release : 'a t -> 'a -> unit } -> 'a desc
47
+
| Fix : { doc : 'a t_; wrt : _ t_ } -> 'a desc
48
+
49
+
(* a set of (active) parents for a ['a t], used during invalidation *)
50
+
and trace =
51
+
| T0
52
+
| T1 : _ t_ -> trace
53
+
| T2 : _ t_ * _ t_ -> trace
54
+
| T3 : _ t_ * _ t_ * _ t_ -> trace
55
+
| T4 : _ t_ * _ t_ * _ t_ * _ t_ -> trace
56
+
| Tn : { mutable active : int; mutable count : int;
57
+
mutable entries : Any.t t_ array } -> trace
58
+
59
+
(* a set of direct children for a composite document *)
60
+
and trace_idx =
61
+
| I0
62
+
| I1 : { mutable idx : int ;
63
+
obj : 'a t_;
64
+
mutable next : trace_idx } -> trace_idx
65
+
66
+
(* The type system cannot see that t is covariant in its parameter.
67
+
Use the Force to convince it. *)
68
+
and +'a t
69
+
external inj : 'a t_ -> 'a t = "%identity"
70
+
external prj : 'a t -> 'a t_ = "%identity"
71
+
external prj2 : 'a t t -> 'a t_ t_ = "%identity"
72
+
73
+
(* Basic combinators *)
74
+
let return x = inj (Pure x)
75
+
let pure x = inj (Pure x)
76
+
77
+
let is_pure x = match prj x with
78
+
| Pure x -> Some x
79
+
| _ -> None
80
+
81
+
let dummy = Pure (Any.any ())
82
+
83
+
let operator desc =
84
+
Operator { value = Eval_none; trace = T0; desc; trace_idx = I0 ;mutex= Mutex.create () }
85
+
86
+
let map x ~f = inj (
87
+
match prj x with
88
+
| Pure vx -> Pure (f vx)
89
+
| x -> operator (Map (x, f))
90
+
)
91
+
92
+
let map2 x y ~f = inj (
93
+
match prj x, prj y with
94
+
| Pure vx, Pure vy -> Pure (f vx vy)
95
+
| x, y -> operator (Map2 (x, y, f))
96
+
)
97
+
98
+
99
+
let pair x y = inj (
100
+
match prj x, prj y with
101
+
| Pure vx, Pure vy -> Pure (vx, vy)
102
+
| x, y -> operator (Pair (x, y))
103
+
)
104
+
105
+
let app f x = inj (
106
+
match prj f, prj x with
107
+
| Pure vf, Pure vx -> Pure (vf vx)
108
+
| f, x -> operator (App (f, x))
109
+
)
110
+
111
+
let join child = inj (
112
+
match prj2 child with
113
+
| Pure v -> v
114
+
| child -> operator (Join { child; intermediate = None })
115
+
)
116
+
117
+
let bind x ~f = join (map ~f x)
118
+
119
+
let pp_eval_status fmt eval =
120
+
match eval with
121
+
| Eval_none -> Format.fprintf fmt "None"
122
+
| Eval_progress -> Format.fprintf fmt "Progress"
123
+
| Eval_some _ -> Format.fprintf fmt "Some"
124
+
| Eval_invalid_next -> Format.fprintf fmt "Invalid_next"
125
+
126
+
(* Management of trace indices *)
127
+
128
+
let addr oc obj =
129
+
Printf.fprintf oc "0x%08x" (Obj.magic obj : int)
130
+
131
+
let pp_addr fmt obj =
132
+
Format.fprintf fmt "0x%08x" (Obj.magic obj : int)
133
+
134
+
external t_equal : _ t_ -> _ t_ -> bool = "%eq"
135
+
external obj_t : 'a t_ -> Any.t t_ = "%identity"
136
+
137
+
let rec dump_trace_format : type a. Format.formatter -> a t_ -> unit =
138
+
fun fmt obj ->
139
+
match obj with
140
+
| Pure _ -> Format.fprintf fmt "%a: Pure _@." pp_addr obj
141
+
| Operator t ->
142
+
Format.fprintf fmt "%a: Operator _ -> %a@." pp_addr obj dump_trace_aux
143
+
t.trace;
144
+
begin
145
+
match t.trace with
146
+
| T0 -> ()
147
+
| T1 a -> dump_trace_format fmt a
148
+
| T2 (a, b) ->
149
+
dump_trace_format fmt a;
150
+
dump_trace_format fmt b
151
+
| T3 (a, b, c) ->
152
+
dump_trace_format fmt a;
153
+
dump_trace_format fmt b;
154
+
dump_trace_format fmt c
155
+
| T4 (a, b, c, d) ->
156
+
dump_trace_format fmt a;
157
+
dump_trace_format fmt b;
158
+
dump_trace_format fmt c;
159
+
dump_trace_format fmt d
160
+
| Tn t -> Array.iter (dump_trace_format fmt) t.entries
161
+
end
162
+
| Root t ->
163
+
Format.fprintf fmt "%a: Root _@." pp_addr obj
11
164
12
-
type 'a eval =
13
-
| Eval_none
14
-
| Eval_progress
15
-
| Eval_some of 'a
16
-
| Eval_invalid_next
165
+
and dump_trace_aux : type a. Format.formatter -> trace -> unit =
166
+
fun fmt -> function
167
+
| T0 -> Format.fprintf fmt "T0"
168
+
| T1 a -> Format.fprintf fmt "T1 %a" pp_addr a
169
+
| T2 (a, b) -> Format.fprintf fmt "T2 (%a, %a)" pp_addr a pp_addr b
170
+
| T3 (a, b, c) ->
171
+
Format.fprintf fmt "T3 (%a, %a, %a)" pp_addr a pp_addr b pp_addr c
172
+
| T4 (a, b, c, d) ->
173
+
Format.fprintf fmt "T4 (%a, %a, %a, %a)" pp_addr a pp_addr b pp_addr c
174
+
pp_addr d
175
+
| Tn t ->
176
+
Format.fprintf fmt "Tn {active = %d; count = %d; entries = " t.active
177
+
t.count;
178
+
Array.iter (Format.fprintf fmt "(%a)" pp_addr) t.entries;
179
+
Format.fprintf fmt "}"
180
+
181
+
let dump_trace x = dump_trace_format Format.err_formatter (obj_t (prj x))
182
+
183
+
let to_mermaid (type a) ?(max_nodes=100) (root : a t_) : string =
184
+
let buf = Buffer.create 1024 in
185
+
Buffer.add_string buf "graph TD;\n";
186
+
let visited : (string, unit) Hashtbl.t = Hashtbl.create 16 in
17
187
18
-
type 'a t_ =
19
-
| Pure of 'a
20
-
| Operator : {
21
-
mutex : Mutex.t;
22
-
mutable value : 'a eval; (* cached value *)
23
-
mutable trace : trace; (* list of parents this can invalidate *)
24
-
mutable trace_idx : trace_idx; (* list of direct children that can invalidate this *)
25
-
desc: 'a desc;
26
-
} -> 'a t_
27
-
| Root : {
28
-
mutex : Mutex.t;
29
-
mutable value : 'a eval; (* cached value *)
30
-
mutable trace_idx : trace_idx; (* list of direct children that can invalidate this *)
31
-
mutable on_invalidate : 'a -> unit;
32
-
mutable acquired : bool;
33
-
child : 'a t_;
34
-
} -> 'a t_
188
+
let get_id : type a. a t_ -> string = fun node -> Format.asprintf "%a" pp_addr node in
35
189
36
-
and _ desc =
37
-
| Map : 'a t_ * ('a -> 'b) -> 'b desc
38
-
| Map2 : 'a t_ * 'b t_ * ('a -> 'b -> 'c) -> 'c desc
39
-
| Pair : 'a t_ * 'b t_ -> ('a * 'b) desc
40
-
| App : ('a -> 'b) t_ * 'a t_ -> 'b desc
41
-
| Join : { child : 'a t_ t_; mutable intermediate : 'a t_ option } -> 'a desc
42
-
| Var : { mutable binding : 'a; mutable nextVal: 'a option } -> 'a desc
43
-
| Prim : { acquire : 'a t -> 'a;
44
-
release : 'a t -> 'a -> unit } -> 'a desc
45
-
| Fix : { doc : 'a t_; wrt : _ t_ } -> 'a desc
190
+
let get_eval_status_str (type a) (eval : a eval) : string =
191
+
match eval with
192
+
| Eval_none -> "None"
193
+
| Eval_progress -> "Progress"
194
+
| Eval_some _ -> "Some"
195
+
| Eval_invalid_next -> "Invalid_next"
196
+
in
46
197
47
-
(* a set of (active) parents for a ['a t], used during invalidation *)
48
-
and trace =
49
-
| T0
50
-
| T1 : _ t_ -> trace
51
-
| T2 : _ t_ * _ t_ -> trace
52
-
| T3 : _ t_ * _ t_ * _ t_ -> trace
53
-
| T4 : _ t_ * _ t_ * _ t_ * _ t_ -> trace
54
-
| Tn : { mutable active : int; mutable count : int;
55
-
mutable entries : Any.t t_ array } -> trace
198
+
(* Using breadth-first traversal with a queue to prevent stack overflows on deep graphs. *)
199
+
let q : Any.t t_ Queue.t = Queue.create () in
200
+
Queue.add (obj_t ( root)) q;
201
+
let nodes_processed = ref 0 in
56
202
57
-
(* a set of direct children for a composite document *)
58
-
and trace_idx =
59
-
| I0
60
-
| I1 : { mutable idx : int ;
61
-
obj : 'a t_;
62
-
mutable next : trace_idx } -> trace_idx
203
+
let rec process_queue () =
204
+
if not (Queue.is_empty q) && !nodes_processed < max_nodes then (
205
+
let node = Queue.take q in
206
+
let node_id_str = get_id node in
207
+
if not (Hashtbl.mem visited node_id_str) then (
208
+
Hashtbl.add visited node_id_str ();
209
+
incr nodes_processed;
210
+
let node_id_str = get_id node in
63
211
64
-
(* The type system cannot see that t is covariant in its parameter.
65
-
Use the Force to convince it. *)
66
-
and +'a t
67
-
external inj : 'a t_ -> 'a t = "%identity"
68
-
external prj : 'a t -> 'a t_ = "%identity"
69
-
external prj2 : 'a t t -> 'a t_ t_ = "%identity"
212
+
let label, children =
213
+
match node with
214
+
| Pure _ -> ("Pure", [])
215
+
| Root t ->
216
+
let status = get_eval_status_str t.value in
217
+
(Printf.sprintf "Root\nstatus: %s" status, [ (obj_t t.child, "solid") ])
218
+
| Operator t ->
219
+
let status = get_eval_status_str t.value in
220
+
let desc_str, children =
221
+
match t.desc with
222
+
| Map (x, _) -> "Map", [ (obj_t x, "solid") ]
223
+
| Map2 (x, y, _) -> "Map2", [ (obj_t x, "solid"); (obj_t y, "solid") ]
224
+
| Pair (x, y) -> "Pair", [ (obj_t x, "solid"); (obj_t y, "solid") ]
225
+
| App (f, x) -> "App", [ (obj_t f, "solid"); (obj_t x, "solid") ]
226
+
| Join { child; intermediate } ->
227
+
let c = [ (obj_t child, "solid") ] in
228
+
let c =
229
+
match intermediate with
230
+
| None -> c
231
+
| Some i -> (obj_t i, "dotted") :: c
232
+
in
233
+
"Join", c
234
+
| Var v ->
235
+
(Printf.sprintf "Var\nhash: %d" (Hashtbl.hash v.binding), [])
236
+
| Prim _ -> "Prim", []
237
+
| Fix { doc; wrt } -> "Fix", [ (obj_t doc, "solid"); (obj_t wrt, "solid") ]
238
+
in
239
+
(Printf.sprintf "%s\nstatus: %s" desc_str status, children)
240
+
in
70
241
71
-
(* Basic combinators *)
72
-
let return x = inj (Pure x)
73
-
let pure x = inj (Pure x)
242
+
let label = Printf.sprintf "%s\n%s" node_id_str label in
243
+
Printf.bprintf buf " %s[\"%s\"];\n" node_id_str label;
74
244
75
-
let is_pure x = match prj x with
76
-
| Pure x -> Some x
77
-
| _ -> None
245
+
if !nodes_processed < max_nodes then (
246
+
List.iter (fun (child, style) ->
247
+
let child_id_str = get_id child in
248
+
let arrow = if style = "dotted" then "-.-> " else "-->" in
249
+
Printf.bprintf buf " %s %s %s;\n" node_id_str arrow child_id_str;
250
+
Queue.add child q
251
+
) children
252
+
) else if children <> [] then (
253
+
let ellipsis_id = node_id_str ^ "_ellipsis" in
254
+
Printf.bprintf buf " %s[\"...\"];\n" ellipsis_id;
255
+
Printf.bprintf buf " %s --> %s;\n" node_id_str ellipsis_id
256
+
);
257
+
);
258
+
process_queue ()
259
+
)
260
+
in
261
+
process_queue ();
262
+
Buffer.contents buf
78
263
79
-
let dummy = Pure (Any.any ())
264
+
let to_mermaid_trace (type a) ?(max_nodes=100) (start : a t_) : string =
265
+
let buf = Buffer.create 1024 in
266
+
Buffer.add_string buf "graph TD;\n";
267
+
let visited : (Any.t t_, unit) Hashtbl.t = Hashtbl.create 16 in
80
268
81
-
let operator desc =
82
-
Operator { value = Eval_none; trace = T0; desc; trace_idx = I0 ;mutex= Mutex.create () }
269
+
let get_id : type a. a t_ -> string = fun node -> Format.asprintf "%a" pp_addr node in
83
270
84
-
let map x ~f = inj (
85
-
match prj x with
86
-
| Pure vx -> Pure (f vx)
87
-
| x -> operator (Map (x, f))
88
-
)
271
+
let get_eval_status_str (type a) (eval : a eval) : string =
272
+
match eval with
273
+
| Eval_none -> "None"
274
+
| Eval_progress -> "Progress"
275
+
| Eval_some _ -> "Some"
276
+
| Eval_invalid_next -> "Invalid_next"
277
+
in
89
278
90
-
let map2 x y ~f = inj (
91
-
match prj x, prj y with
92
-
| Pure vx, Pure vy -> Pure (f vx vy)
93
-
| x, y -> operator (Map2 (x, y, f))
94
-
)
279
+
(* Using breadth-first traversal following trace (parent) relationships *)
280
+
let q : Any.t t_ Queue.t = Queue.create () in
281
+
Queue.add (obj_t ( start)) q;
282
+
let nodes_processed = ref 0 in
95
283
284
+
let rec process_queue () =
285
+
if not (Queue.is_empty q) && !nodes_processed < max_nodes then (
286
+
let node = Queue.take q in
287
+
if not (Hashtbl.mem visited node) then (
288
+
Hashtbl.add visited node ();
289
+
incr nodes_processed;
290
+
let node_id_str = get_id node in
96
291
97
-
let pair x y = inj (
98
-
match prj x, prj y with
99
-
| Pure vx, Pure vy -> Pure (vx, vy)
100
-
| x, y -> operator (Pair (x, y))
101
-
)
292
+
let label, parents =
293
+
match node with
294
+
| Pure _ -> ("Pure", [])
295
+
| Root t ->
296
+
let status = get_eval_status_str t.value in
297
+
(Printf.sprintf "Root\nstatus: %s" status, [])
298
+
| Operator t ->
299
+
let status = get_eval_status_str t.value in
300
+
let desc_str =
301
+
match t.desc with
302
+
| Map (_, _) -> "Map"
303
+
| Map2 (_, _, _) -> "Map2"
304
+
| Pair (_, _) -> "Pair"
305
+
| App (_, _) -> "App"
306
+
| Join _ -> "Join"
307
+
| Var v -> Printf.sprintf "Var\nhash: %d" (Hashtbl.hash v.binding)
308
+
| Prim _ -> "Prim"
309
+
| Fix _ -> "Fix"
310
+
in
311
+
let parents =
312
+
match t.trace with
313
+
| T0 -> []
314
+
| T1 p1 -> [obj_t p1]
315
+
| T2 (p1, p2) -> [obj_t p1; obj_t p2]
316
+
| T3 (p1, p2, p3) -> [obj_t p1; obj_t p2; obj_t p3]
317
+
| T4 (p1, p2, p3, p4) -> [obj_t p1; obj_t p2; obj_t p3; obj_t p4]
318
+
| Tn t -> Array.to_list (Array.sub t.entries 0 t.active)
319
+
in
320
+
(Printf.sprintf "%s\nstatus: %s" desc_str status, parents)
321
+
in
102
322
103
-
let app f x = inj (
104
-
match prj f, prj x with
105
-
| Pure vf, Pure vx -> Pure (vf vx)
106
-
| f, x -> operator (App (f, x))
107
-
)
323
+
let label = Printf.sprintf "%s\n%s" node_id_str label in
324
+
Printf.bprintf buf " %s[\"%s\"];\n" node_id_str label;
108
325
109
-
let join child = inj (
110
-
match prj2 child with
111
-
| Pure v -> v
112
-
| child -> operator (Join { child; intermediate = None })
113
-
)
326
+
if !nodes_processed < max_nodes then (
327
+
List.iter (fun parent ->
328
+
let parent_id_str = get_id parent in
329
+
Printf.bprintf buf " %s --> %s;\n" parent_id_str node_id_str;
330
+
Queue.add parent q
331
+
) parents
332
+
) else if parents <> [] then (
333
+
let ellipsis_id = node_id_str ^ "_ellipsis" in
334
+
Printf.bprintf buf " %s[\"...\"];\n" ellipsis_id;
335
+
Printf.bprintf buf " %s --> %s;\n" ellipsis_id node_id_str
336
+
);
337
+
);
338
+
process_queue ()
339
+
)
340
+
in
341
+
process_queue ();
342
+
Buffer.contents buf
114
343
115
-
let bind x ~f = join (map ~f x)
344
+
let to_mermaid_trace_idx (type a) ?(max_nodes=100) (start : a t_) : string =
345
+
let buf = Buffer.create 1024 in
346
+
Buffer.add_string buf "graph TD;\n";
347
+
let visited : (Any.t t_, unit) Hashtbl.t = Hashtbl.create 16 in
116
348
117
-
(* Management of trace indices *)
349
+
let get_id : type a. a t_ -> string = fun node -> Format.asprintf "%a" pp_addr node in
118
350
119
-
let addr oc obj =
120
-
Printf.fprintf oc "0x%08x" (Obj.magic obj : int)
351
+
let get_eval_status_str (type a) (eval : a eval) : string =
352
+
match eval with
353
+
| Eval_none -> "None"
354
+
| Eval_progress -> "Progress"
355
+
| Eval_some _ -> "Some"
356
+
| Eval_invalid_next -> "Invalid_next"
357
+
in
121
358
122
-
external t_equal : _ t_ -> _ t_ -> bool = "%eq"
123
-
external obj_t : 'a t_ -> Any.t t_ = "%identity"
359
+
(* Using breadth-first traversal with a queue to prevent stack overflows on deep graphs. *)
360
+
let q : Any.t t_ Queue.t = Queue.create () in
361
+
Queue.add (obj_t (start)) q;
362
+
let nodes_processed = ref 0 in
124
363
125
-
let rec dump_trace : type a. a t_ -> unit =
126
-
fun obj -> match obj with
127
-
| Pure _ -> Printf.eprintf "%a: Pure _\n%!" addr obj
128
-
| Operator t ->
129
-
Printf.eprintf "%a: Operator _ -> %a\n%!" addr obj dump_trace_aux t.trace;
130
-
begin match t.trace with
131
-
| T0 -> ()
132
-
| T1 a -> dump_trace a
133
-
| T2 (a,b) -> dump_trace a; dump_trace b
134
-
| T3 (a,b,c) -> dump_trace a; dump_trace b; dump_trace c
135
-
| T4 (a,b,c,d) -> dump_trace a; dump_trace b; dump_trace c; dump_trace d
136
-
| Tn t -> Array.iter dump_trace t.entries
137
-
end
138
-
| Root _ -> Printf.eprintf "%a: Root _\n%!" addr obj
364
+
let rec process_queue () =
365
+
if not (Queue.is_empty q) && !nodes_processed < max_nodes then (
366
+
let node = Queue.take q in
367
+
if not (Hashtbl.mem visited node) then (
368
+
Hashtbl.add visited node ();
369
+
incr nodes_processed;
370
+
let node_id_str = get_id node in
139
371
140
-
and dump_trace_aux oc = function
141
-
| T0 -> Printf.fprintf oc "T0"
142
-
| T1 a -> Printf.fprintf oc "T1 %a" addr a
143
-
| T2 (a,b) ->
144
-
Printf.fprintf oc "T2 (%a, %a)" addr a addr b
145
-
| T3 (a,b,c) ->
146
-
Printf.fprintf oc "T3 (%a, %a, %a)" addr a addr b addr c
147
-
| T4 (a,b,c,d) ->
148
-
Printf.fprintf oc "T4 (%a, %a, %a, %a)" addr a addr b addr c addr d
149
-
| Tn t ->
150
-
Printf.fprintf oc "Tn {active = %d; count = %d; entries = "
151
-
t.active t.count;
152
-
Array.iter (Printf.fprintf oc "(%a)" addr) t.entries;
153
-
Printf.fprintf oc "}"
154
-
155
-
let dump_trace x = dump_trace (obj_t (prj x))
156
-
157
-
let add_idx obj idx = function
158
-
| Pure _ -> assert false
159
-
| Root t' -> t'.trace_idx <- I1 { idx; obj; next = t'.trace_idx }
160
-
| Operator t' -> t'.trace_idx <- I1 { idx; obj; next = t'.trace_idx }
161
-
162
-
let rec rem_idx_rec obj = function
163
-
| I0 -> assert false
164
-
| I1 t as self ->
165
-
if t_equal t.obj obj
166
-
then (t.idx, t.next)
167
-
else (
168
-
let idx, result = rem_idx_rec obj t.next in
169
-
t.next <- result;
170
-
(idx, self)
171
-
)
172
-
173
-
(* remove [obj] from the lwd's trace. *)
174
-
let rem_idx obj = function
175
-
| Pure _ -> assert false
176
-
| Root t' ->
177
-
let idx, trace_idx = rem_idx_rec obj t'.trace_idx in
178
-
t'.trace_idx <- trace_idx; idx
179
-
| Operator t' ->
180
-
let idx, trace_idx = rem_idx_rec obj t'.trace_idx in
181
-
t'.trace_idx <- trace_idx; idx
182
-
183
-
(* move [obj] from old index to new index. *)
184
-
let rec mov_idx_rec obj oldidx newidx = function
185
-
| I0 -> assert false
186
-
| I1 t ->
187
-
if t.idx = oldidx && t_equal t.obj obj
188
-
then t.idx <- newidx
189
-
else mov_idx_rec obj oldidx newidx t.next
190
-
191
-
let mov_idx obj oldidx newidx = function
192
-
| Pure _ -> assert false
193
-
| Root t' -> mov_idx_rec obj oldidx newidx t'.trace_idx
194
-
| Operator t' -> mov_idx_rec obj oldidx newidx t'.trace_idx
195
-
196
-
let rec get_idx_rec obj = function
197
-
| I0 -> assert false
198
-
| I1 t ->
199
-
if t_equal t.obj obj
200
-
then t.idx
201
-
else get_idx_rec obj t.next
372
+
let get_children_from_trace_idx trace_idx =
373
+
let rec aux acc = function
374
+
| I0 -> acc
375
+
| I1 { obj; next; _ } -> aux ((obj_t obj, "solid") :: acc) next
376
+
in
377
+
aux [] trace_idx
378
+
in
202
379
203
-
(* find index of [obj] in the given lwd *)
204
-
let get_idx obj = function
205
-
| Pure _ -> assert false
206
-
| Root t' -> get_idx_rec obj t'.trace_idx
207
-
| Operator t' -> get_idx_rec obj t'.trace_idx
208
-
209
-
type status =
210
-
| Neutral
211
-
| Safe
212
-
| Unsafe of (unit->unit) list ref
380
+
let label, children =
381
+
match node with
382
+
| Pure _ -> ("Pure", [])
383
+
| Root t ->
384
+
let status = get_eval_status_str t.value in
385
+
let children = get_children_from_trace_idx t.trace_idx in
386
+
(Printf.sprintf "Root\nstatus: %s" status, children)
387
+
| Operator t ->
388
+
let status = get_eval_status_str t.value in
389
+
let desc_str =
390
+
match t.desc with
391
+
| Map (_, _) -> "Map"
392
+
| Map2 (_, _, _) -> "Map2"
393
+
| Pair (_, _) -> "Pair"
394
+
| App (_, _) -> "App"
395
+
| Join _ -> "Join"
396
+
| Var v -> Printf.sprintf "Var\nhash: %d" (Hashtbl.hash v.binding)
397
+
| Prim _ -> "Prim"
398
+
| Fix _ -> "Fix"
399
+
in
400
+
let children = get_children_from_trace_idx t.trace_idx in
401
+
(Printf.sprintf "%s\nstatus: %s" desc_str status, children)
402
+
in
213
403
214
-
(*
215
-
Sensitivity is used to indicate to when reading a root node, that one of the child operater nodes was being evaluated.
216
-
I think this is needed because the child cound have multiple roots and we need to indicate that to all of them
217
-
*)
218
-
type sensitivity =
219
-
| Strong
220
-
| Fragile
404
+
let label = Printf.sprintf "%s\n%s" node_id_str label in
405
+
Printf.bprintf buf " %s[\"%s\"];\n" node_id_str label;
221
406
222
-
(* Propagating invalidation recursively.
223
-
Each document is invalidated at most once,
224
-
and only if it has [t.value = Some _]. *)
225
-
let rec invalidate_node : type a . status ref -> sensitivity -> a t_ -> unit =
226
-
(*sensitivity indicates that a parent is being evaluated*)
227
-
fun status sensitivity node ->
228
-
match node, sensitivity with
229
-
| Pure _, _ -> assert false
230
-
| Root ({value; on_invalidate; _} as t), _ ->
231
-
(match value with
232
-
| Eval_none | Eval_invalid_next -> ()
233
-
| Eval_progress ->
234
-
t.value <- Eval_invalid_next
235
-
| Eval_some x ->
236
-
t.value <- Eval_none;
237
-
on_invalidate x
407
+
if !nodes_processed < max_nodes then (
408
+
List.iter (fun (child, style) ->
409
+
let child_id_str = get_id child in
410
+
let arrow = if style = "dotted" then "-.-> " else "-->" in
411
+
Printf.bprintf buf " %s %s %s;\n" node_id_str arrow child_id_str;
412
+
Queue.add child q
413
+
) children
414
+
) else if children <> [] then (
415
+
let ellipsis_id = node_id_str ^ "_ellipsis" in
416
+
Printf.bprintf buf " %s[\"...\"];\n" ellipsis_id;
417
+
Printf.bprintf buf " %s --> %s;\n" node_id_str ellipsis_id
418
+
);
419
+
);
420
+
process_queue ()
238
421
)
239
-
| Operator { value = Eval_none | Eval_invalid_next; _ }, _ -> ()
240
-
| Operator { desc = Fix { wrt = Operator { value = Eval_none | Eval_invalid_next; _ }; _ }; _ }, Fragile ->
241
-
(match !status with
242
-
| Safe | Unsafe _ -> ()
243
-
| Neutral -> status := Safe)
244
-
| Operator { desc = Fix { wrt = Operator { value = Eval_some _; _ }; _ }; _ }, Fragile
245
-
-> ()
246
-
| Operator t, _ ->
247
-
let sensitivity =
248
-
match t.value with Eval_progress -> Fragile | _ -> sensitivity
249
422
in
250
-
t.value <- Eval_none;
251
-
(* invalidate parents recursively *)
252
-
invalidate_trace status sensitivity t.trace
423
+
process_queue ();
424
+
Buffer.contents buf
253
425
254
-
(* invalidate recursively documents in the given trace *)
255
-
and invalidate_trace status sensitivity = function
256
-
| T0 -> ()
257
-
| T1 x -> invalidate_node status sensitivity x
258
-
| T2 (x, y) ->
259
-
invalidate_node status sensitivity x;
260
-
invalidate_node status sensitivity y
261
-
| T3 (x, y, z) ->
262
-
invalidate_node status sensitivity x;
263
-
invalidate_node status sensitivity y;
264
-
invalidate_node status sensitivity z
265
-
| T4 (x, y, z, w) ->
266
-
invalidate_node status sensitivity x;
267
-
invalidate_node status sensitivity y;
268
-
invalidate_node status sensitivity z;
269
-
invalidate_node status sensitivity w
270
-
| Tn t ->
271
-
let active = t.active in
272
-
t.active <- 0;
273
-
for i = 0 to active - 1 do
274
-
invalidate_node status sensitivity t.entries.(i)
275
-
done
426
+
let add_idx obj idx = function
427
+
| Pure _ -> assert false
428
+
| Root t' -> t'.trace_idx <- I1 { idx; obj; next = t'.trace_idx }
429
+
| Operator t' -> t'.trace_idx <- I1 { idx; obj; next = t'.trace_idx }
430
+
431
+
let rec rem_idx_rec obj = function
432
+
| I0 -> assert false
433
+
| I1 t as self ->
434
+
if t_equal t.obj obj
435
+
then (t.idx, t.next)
436
+
else (
437
+
let idx, result = rem_idx_rec obj t.next in
438
+
t.next <- result;
439
+
(idx, self)
440
+
)
441
+
442
+
(* remove [obj] from the lwd's trace. *)
443
+
let rem_idx obj = function
444
+
| Pure _ -> assert false
445
+
| Root t' ->
446
+
let idx, trace_idx = rem_idx_rec obj t'.trace_idx in
447
+
t'.trace_idx <- trace_idx; idx
448
+
| Operator t' ->
449
+
let idx, trace_idx = rem_idx_rec obj t'.trace_idx in
450
+
t'.trace_idx <- trace_idx; idx
451
+
452
+
(* move [obj] from old index to new index. *)
453
+
let rec mov_idx_rec obj oldidx newidx = function
454
+
| I0 -> assert false
455
+
| I1 t ->
456
+
if t.idx = oldidx && t_equal t.obj obj
457
+
then t.idx <- newidx
458
+
else mov_idx_rec obj oldidx newidx t.next
459
+
460
+
let mov_idx obj oldidx newidx = function
461
+
| Pure _ -> assert false
462
+
| Root t' -> mov_idx_rec obj oldidx newidx t'.trace_idx
463
+
| Operator t' -> mov_idx_rec obj oldidx newidx t'.trace_idx
464
+
465
+
let rec get_idx_rec obj = function
466
+
| I0 -> assert false
467
+
| I1 t ->
468
+
if t_equal t.obj obj
469
+
then t.idx
470
+
else get_idx_rec obj t.next
471
+
472
+
(* find index of [obj] in the given lwd *)
473
+
let get_idx obj = function
474
+
| Pure _ -> assert false
475
+
| Root t' -> get_idx_rec obj t'.trace_idx
476
+
| Operator t' -> get_idx_rec obj t'.trace_idx
477
+
478
+
type status =
479
+
| Neutral
480
+
| Safe
481
+
| Unsafe of (unit->unit) list ref
482
+
483
+
(*
484
+
Sensitivity is used to indicate to when reading a root node, that one of the child operater nodes was being evaluated.
485
+
I think this is needed because the child cound have multiple roots and we need to indicate that to all of them
486
+
*)
487
+
type sensitivity =
488
+
| Strong
489
+
| Fragile
490
+
491
+
let pp_sensitivity ppf = function
492
+
| Strong -> Format.fprintf ppf "Strong"
493
+
| Fragile -> Format.fprintf ppf "Fragile"
494
+
495
+
(* Propagating invalidation recursively.
496
+
Each document is invalidated at most once,
497
+
and only if it has [t.value = Some _]. *)
498
+
let rec invalidate_node : type a . status ref -> sensitivity -> a t_ -> unit =
499
+
(*sensitivity indicates that a parent is being evaluated*)
500
+
fun status sensitivity node ->
276
501
277
-
let default_unsafe_mutation_logger () =
278
-
let callstack = Printexc.get_callstack 20 in
279
-
Printf.fprintf stderr
280
-
"Lwd: unsafe mutation (variable invalidated during evaluation) at\n%a"
281
-
Printexc.print_raw_backtrace callstack
502
+
match node, sensitivity with
503
+
| Pure _, _ -> assert false
504
+
| Root ({value; on_invalidate; _} as t), _ ->
505
+
(match value with
506
+
| Eval_none | Eval_invalid_next -> ()
507
+
| Eval_progress ->
508
+
t.value <- Eval_invalid_next
509
+
| Eval_some x ->
510
+
t.value <- Eval_none;
511
+
on_invalidate x
512
+
)
513
+
| Operator { value = Eval_none | Eval_invalid_next; _ }, _ -> ()
514
+
| Operator { desc = Fix { wrt = Operator { value = Eval_none | Eval_invalid_next; _ }; _ }; _ }, Fragile ->
515
+
(match !status with
516
+
| Safe | Unsafe _ -> ()
517
+
| Neutral -> status := Safe)
518
+
| Operator { desc = Fix { wrt = Operator { value = Eval_some _; _ }; _ }; _ }, Fragile
519
+
-> ()
520
+
| Operator t, _ ->
521
+
let sensitivity =
522
+
match t.value with Eval_progress -> Fragile | _ -> sensitivity
523
+
in
524
+
t.value <- Eval_none;
525
+
(* invalidate parents recursively *)
526
+
invalidate_trace status sensitivity t.trace
527
+
528
+
(* invalidate recursively documents in the given trace *)
529
+
and invalidate_trace status sensitivity = function
530
+
| T0 -> ()
531
+
| T1 x -> invalidate_node status sensitivity x
532
+
| T2 (x, y) ->
533
+
invalidate_node status sensitivity x;
534
+
invalidate_node status sensitivity y
535
+
| T3 (x, y, z) ->
536
+
invalidate_node status sensitivity x;
537
+
invalidate_node status sensitivity y;
538
+
invalidate_node status sensitivity z
539
+
| T4 (x, y, z, w) ->
540
+
invalidate_node status sensitivity x;
541
+
invalidate_node status sensitivity y;
542
+
invalidate_node status sensitivity z;
543
+
invalidate_node status sensitivity w
544
+
| Tn t ->
545
+
let active = t.active in
546
+
t.active <- 0;
547
+
for i = 0 to active - 1 do
548
+
invalidate_node status sensitivity t.entries.(i)
549
+
done
550
+
551
+
let default_unsafe_mutation_logger () =
552
+
let callstack = Printexc.get_callstack 20 in
553
+
Printf.fprintf stderr
554
+
"Lwd: unsafe mutation (variable invalidated during evaluation) at\n%a"
555
+
Printexc.print_raw_backtrace callstack
556
+
557
+
let unsafe_mutation_logger = ref default_unsafe_mutation_logger
558
+
559
+
560
+
let do_invalidate sensitivity (node : 'a t_) =
561
+
let status = ref Neutral in
562
+
invalidate_node status sensitivity node;
563
+
(* Variables *)
564
+
type 'a var = 'a t_
565
+
let var x = operator (Var {binding = x})
566
+
let get x = inj x
567
+
568
+
let get_parents_from_trace (trace:trace) : (Any.t var) list =
569
+
match trace with
570
+
| T0 -> []
571
+
| T1 p1 -> [obj_t p1]
572
+
| T2 (p1, p2) -> [obj_t p1; obj_t p2]
573
+
| T3 (p1, p2, p3) -> [obj_t p1; obj_t p2; obj_t p3]
574
+
| T4 (p1, p2, p3, p4) -> [obj_t p1; obj_t p2; obj_t p3; obj_t p4]
575
+
| Tn t ->
576
+
let res = ref [] in
577
+
for i = t.active - 1 downto 0 do
578
+
res := t.entries.(i) :: !res
579
+
done;
580
+
!res
282
581
283
-
let unsafe_mutation_logger = ref default_unsafe_mutation_logger
582
+
let set (vx:'a var) x : unit =
583
+
match vx with
584
+
| (Operator ({desc = Var v; _;} as inner )) ->
585
+
v.binding <- x;
586
+
inner.value <- Eval_some x;
587
+
(* [climb] traverses the graph upwards from a changed variable, invalidating
588
+
parent nodes. The `parents` list is a list of work to do, containing tuples
589
+
of `(node, has_seen_invalid_next)`.
284
590
591
+
The `has_seen_invalid_next` flag is crucial for correctness. It tracks
592
+
whether we have already encountered a node marked `Eval_invalid_next`
593
+
in the current upward traversal path.
285
594
286
-
(**
287
-
595
+
When a node is being evaluated (`Eval_progress`), and a variable it
596
+
depends on is `set`, the evaluation might be happening in a different
597
+
thread. If another invalidation has already marked a node higher up in
598
+
the chain as `Eval_invalid_next`, we don't want to also mark the
599
+
current `Eval_progress` node as `Eval_invalid_next`. The higher-up
600
+
invalidation will already cause a re-evaluation that will deal with
601
+
this node. Setting it to `Eval_invalid_next` here could lead to
602
+
incorrect state transitions.
288
603
289
-
@param ~was_delayed: set to true if the function call was put on hold untill the current root had finished being evaluated
290
-
*)
291
-
let do_invalidate sensitivity (node : 'a t_) =
292
-
let status = ref Neutral in
293
-
invalidate_node status sensitivity node;
294
-
(* Variables *)
295
-
type 'a var = 'a t_
296
-
let var x = operator (Var {binding = x;nextVal=None})
297
-
let get x = inj x
298
-
299
-
let set (vx:'a var) x : unit =
300
-
match vx with
301
-
| (Operator ({desc = Var v; _} )as node) ->
302
-
v.binding <- x;
303
-
v.nextVal <- None;
304
-
(let rec climb (parents: Any.t t_ list) =
305
-
let new_parents : Any.t t_ list = List.fold_left (fun acc p ->
306
-
match p with
307
-
| Pure _ -> acc
308
-
| Root r ->
309
-
if Mutex.try_lock r.mutex then (
310
-
(match r.value with
311
-
| Eval_some v ->
312
-
r.value <- Eval_none;
313
-
r.on_invalidate v
314
-
| Eval_none | Eval_invalid_next -> ()
315
-
| Eval_progress -> r.value <- Eval_invalid_next
316
-
);
317
-
Mutex.unlock r.mutex;
318
-
acc
319
-
) else (
320
-
Mutex.protect r.mutex (fun () ->
321
-
if r.value = Eval_progress then r.value <- Eval_invalid_next
604
+
Therefore, `has_seen_invalid_next` is propagated upwards. Once it's
605
+
`true` for a given path, it remains `true` for all ancestors in that
606
+
path. We only set a node from `Eval_progress` to `Eval_invalid_next`
607
+
if `has_seen_invalid_next` is `false`.
608
+
We should actually be able to stop propagating invalidation once we have seen an Eval_invalid_next node, but in reality things seem to be messier than that and so we keep going.
609
+
Instead i reset the seen_eval node flag if we ever encounter a node that is not Eval_progress.
610
+
*)
611
+
(let rec climb (parents: (Any.t t_ * bool) list) =
612
+
let new_parents : (Any.t t_ * bool) list = List.fold_left (fun acc (p, seen_eval_node) ->
613
+
match p with
614
+
| Pure _ -> acc
615
+
| Root r ->
616
+
if Mutex.try_lock r.mutex then (
617
+
(match r.value with
618
+
| Eval_some v ->
619
+
r.value <- Eval_none;
620
+
r.on_invalidate v
621
+
| Eval_none | Eval_invalid_next -> ()
622
+
| Eval_progress -> if not seen_eval_node then r.value <- Eval_invalid_next
322
623
);
323
-
acc
324
-
)
325
-
| Operator o ->
326
-
if Mutex.try_lock o.mutex then (
327
-
let continue =
328
-
match o.value with
329
-
| Eval_some _ -> o.value <- Eval_none; true
330
-
| Eval_none | Eval_invalid_next -> false
331
-
| Eval_progress -> o.value <- Eval_invalid_next; false
624
+
Mutex.unlock r.mutex;
625
+
acc
626
+
) else (
627
+
(* if the root is currently being evaluated, we pro*)
628
+
(* Mutex.protect r.mutex (fun () -> *)
629
+
(* ); *)
630
+
(* ELI: try just skipping the lock and invalidating the root, it should be safe *)
631
+
if r.value = Eval_progress && not seen_eval_node then r.value <- Eval_invalid_next;
632
+
acc
633
+
)
634
+
| Operator o ->
635
+
let (continue, seen_eval_node) =
636
+
if Mutex.try_lock o.mutex then (
637
+
let current_value = o.value in
638
+
let continue,this_node_has_seen_invalid_next =
639
+
match current_value with
640
+
| Eval_some _ -> o.value <- Eval_none; true,false
641
+
(* This shouldn't be needed, but sometimes it is so we do it anyway*)
642
+
| Eval_none -> true,false
643
+
| Eval_invalid_next -> true,true
644
+
| Eval_progress -> (if not seen_eval_node then
645
+
o.value <- Eval_invalid_next); true,true
646
+
in
647
+
Mutex.unlock o.mutex;
648
+
(continue, this_node_has_seen_invalid_next )
649
+
) else (
650
+
Mutex.protect o.mutex (fun () ->
651
+
if o.value = Eval_progress && not seen_eval_node then o.value <- Eval_invalid_next
652
+
);
653
+
(true, true)
654
+
)
332
655
in
333
-
Mutex.unlock o.mutex;
334
656
if continue then (
335
-
match o.trace with
336
-
| T0 -> acc
337
-
| T1 p1 -> obj_t p1 :: acc
338
-
| T2 (p1, p2) -> obj_t p1 :: obj_t p2 :: acc
339
-
| T3 (p1, p2, p3) -> obj_t p1 :: obj_t p2 :: obj_t p3 :: acc
340
-
| T4 (p1, p2, p3, p4) -> obj_t p1 :: obj_t p2 :: obj_t p3 :: obj_t p4 :: acc
341
-
| Tn t -> Array.to_list t.entries |> List.rev_append acc
657
+
let parents_of_o = get_parents_from_trace o.trace in
658
+
let new_acc_entries = List.map (fun p -> (obj_t p, seen_eval_node)) parents_of_o in
659
+
List.rev_append new_acc_entries acc
342
660
) else acc
343
-
) else (
344
-
Mutex.protect o.mutex (fun () ->
345
-
if o.value = Eval_progress then o.value <- Eval_invalid_next
346
-
);
347
-
acc
348
-
)
349
-
) [] parents in
350
-
if new_parents <> [] then climb new_parents
351
-
in
352
-
match node with
353
-
| Operator o ->
354
-
let initial_parents : Any.t t_ list =
355
-
match o.trace with
356
-
| T0 -> []
357
-
| T1 p1 -> [obj_t p1]
358
-
| T2 (p1, p2) -> [obj_t p1; obj_t p2]
359
-
| T3 (p1, p2, p3) -> [obj_t p1; obj_t p2; obj_t p3]
360
-
| T4 (p1, p2, p3, p4) -> [obj_t p1; obj_t p2; obj_t p3; obj_t p4]
361
-
| Tn t -> Array.to_list t.entries
661
+
) [] parents in
662
+
if new_parents <> [] then climb new_parents
663
+
in
664
+
let initial_parents = get_parents_from_trace inner.trace in
665
+
climb (List.map (fun p -> (obj_t p, false)) initial_parents)
666
+
)
667
+
| _ -> assert false
668
+
669
+
let peek_stable = function
670
+
| Operator ({desc = Var v; _}) -> v.binding
671
+
| _ -> assert false
672
+
673
+
let peek = function
674
+
| Operator ({desc = Var v; _}) -> v.binding
675
+
| _ -> assert false
676
+
677
+
let update f v = set v (f (peek v))
678
+
679
+
let may_update f v =
680
+
match f (peek v) with
681
+
| None -> ()
682
+
| Some x -> set v x
683
+
684
+
(* Primitives *)
685
+
type 'a prim = 'a t
686
+
let prim ~acquire ~release =
687
+
inj (operator (Prim { acquire; release }))
688
+
let get_prim x = x
689
+
690
+
let invalidate x = match prj x with
691
+
| Operator {desc = Prim p; value; _} as t ->
692
+
(* the value is invalidated, be sure to invalidate all parents as well *)
693
+
begin match value with
694
+
| Eval_none | Eval_invalid_next -> ()
695
+
| Eval_progress -> do_invalidate Fragile t;
696
+
| Eval_some v ->
697
+
do_invalidate Strong t;
698
+
p.release x v
699
+
end
700
+
| _ -> assert false
701
+
702
+
(* Fix point *)
703
+
704
+
let fix doc ~wrt = match prj wrt with
705
+
| Root _ -> assert false
706
+
| Pure _ -> doc
707
+
| Operator _ as wrt -> inj (operator (Fix {doc = prj doc; wrt}))
708
+
709
+
type release_list =
710
+
| Release_done
711
+
| Release_more :
712
+
{ origin : 'a t_; element : 'b t_; next : release_list } -> release_list
713
+
714
+
type release_queue = release_list ref
715
+
let make_release_queue () = ref Release_done
716
+
717
+
type release_failure = exn * Printexc.raw_backtrace
718
+
719
+
(* [sub_release [] origin self] is called when [origin] is released,
720
+
where [origin] is reachable from [self]'s trace.
721
+
We're going to remove [origin] from that trace as [origin] is now dead.
722
+
723
+
[sub_release] cannot raise.
724
+
If a primitive raises, the exception is caught and a warning is emitted. *)
725
+
let rec sub_release
726
+
: type a b . release_failure list -> a t_ -> b t_ -> release_failure list
727
+
= fun failures origin -> function
728
+
| Root _ -> assert false
729
+
| Pure _ -> failures
730
+
| Operator t as self ->
731
+
Mutex.protect t.mutex @@ fun () ->
732
+
(* compute [t.trace \ {origin}] *)
733
+
let trace = match t.trace with
734
+
| T0 -> assert false
735
+
| T1 x -> assert (t_equal x origin); T0
736
+
| T2 (x, y) ->
737
+
if t_equal x origin then T1 y
738
+
else if t_equal y origin then T1 x
739
+
else assert false
740
+
| T3 (x, y, z) ->
741
+
if t_equal x origin then T2 (y, z)
742
+
else if t_equal y origin then T2 (x, z)
743
+
else if t_equal z origin then T2 (x, y)
744
+
else assert false
745
+
| T4 (x, y, z, w) ->
746
+
if t_equal x origin then T3 (y, z, w)
747
+
else if t_equal y origin then T3 (x, z, w)
748
+
else if t_equal z origin then T3 (x, y, w)
749
+
else if t_equal w origin then T3 (x, y, z)
750
+
else assert false
751
+
| Tn tn as trace ->
752
+
let revidx = rem_idx self origin in
753
+
assert (t_equal tn.entries.(revidx) origin);
754
+
let count = tn.count - 1 in
755
+
tn.count <- count;
756
+
if revidx < count then (
757
+
let obj = tn.entries.(count) in
758
+
tn.entries.(revidx) <- obj;
759
+
tn.entries.(count) <- dummy;
760
+
mov_idx self count revidx obj
761
+
) else
762
+
tn.entries.(revidx) <- dummy;
763
+
if tn.active > count then tn.active <- count;
764
+
if count = 4 then (
765
+
(* downgrade to [T4] to save space *)
766
+
let a = tn.entries.(0) and b = tn.entries.(1) in
767
+
let c = tn.entries.(2) and d = tn.entries.(3) in
768
+
ignore (rem_idx self a : int);
769
+
ignore (rem_idx self b : int);
770
+
ignore (rem_idx self c : int);
771
+
ignore (rem_idx self d : int);
772
+
T4 (a, b, c, d)
773
+
) else (
774
+
let len = Array.length tn.entries in
775
+
if count <= len lsr 2 then
776
+
Tn { active = tn.active; count = tn.count;
777
+
entries = Array.sub tn.entries 0 (len lsr 1) }
778
+
else
779
+
trace
780
+
)
362
781
in
363
-
climb initial_parents
782
+
t.trace <- trace;
783
+
match trace with
784
+
| T0 ->
785
+
(* [self] is not active anymore, since it's not reachable
786
+
from any root. We can release its cached value and
787
+
recursively release its subtree. *)
788
+
let value = t.value in
789
+
t.value <- Eval_none;
790
+
begin match t.desc with
791
+
| Map (x, _) -> sub_release failures self x
792
+
| Map2 (x, y, _) ->
793
+
sub_release (sub_release failures self x) self y
794
+
| Pair (x, y) ->
795
+
sub_release (sub_release failures self x) self y
796
+
| App (x, y) ->
797
+
sub_release (sub_release failures self x) self y
798
+
| Join ({ child; intermediate } as t) ->
799
+
let failures = sub_release failures self child in
800
+
begin match intermediate with
801
+
| None -> failures
802
+
| Some child' ->
803
+
t.intermediate <- None;
804
+
sub_release failures self child'
805
+
end
806
+
| Var _ -> failures
807
+
| Fix {doc; wrt} ->
808
+
sub_release (sub_release failures self wrt) self doc
809
+
| Prim t ->
810
+
begin match value with
811
+
| Eval_none | Eval_invalid_next | Eval_progress -> failures
812
+
| Eval_some x ->
813
+
begin match t.release (inj self) x with
814
+
| () -> failures
815
+
| exception exn ->
816
+
let bt = Printexc.get_raw_backtrace () in
817
+
(exn, bt) :: failures
818
+
end
819
+
end
820
+
end
821
+
| _ -> failures
822
+
823
+
(* [sub_acquire] cannot raise *)
824
+
let rec sub_acquire : type a b . a t_ -> b t_ -> unit = fun origin ->
825
+
function
826
+
| Root _ -> assert false
827
+
| Pure _ -> ()
828
+
| Operator t as self ->
829
+
(*lock the mutex, because we are making changes within this node *)
830
+
831
+
Mutex.protect t.mutex @@ fun _->
832
+
(* [acquire] is true if this is the first time this operator
833
+
is used, in which case we need to acquire its children *)
834
+
let acquire = match t.trace with T0 -> true | _ -> false in
835
+
let trace = match t.trace with
836
+
| T0 -> T1 origin
837
+
| T1 x -> T2 (origin, x)
838
+
| T2 (x, y) -> T3 (origin, x, y)
839
+
| T3 (x, y, z) -> T4 (origin, x, y, z)
840
+
| T4 (x, y, z, w) ->
841
+
let obj_origin = obj_t origin in
842
+
let entries =
843
+
[| obj_t x; obj_t y; obj_t z; obj_t w; obj_t origin; dummy; dummy; dummy |]
844
+
in
845
+
for i = 0 to 4 do add_idx self i entries.(i) done;
846
+
Tn { active = 5; count = 5; entries }
847
+
| Tn tn as trace ->
848
+
let index = tn.count in
849
+
let entries, trace =
850
+
(* possibly resize array [entries] *)
851
+
if index < Array.length tn.entries then (
852
+
tn.count <- tn.count + 1;
853
+
(tn.entries, trace)
854
+
) else (
855
+
let entries = Array.make (index * 2) dummy in
856
+
Array.blit tn.entries 0 entries 0 index;
857
+
(entries, Tn { active = tn.active; count = index + 1; entries })
858
+
)
859
+
in
860
+
let obj_origin = obj_t origin in
861
+
entries.(index) <- obj_origin;
862
+
add_idx self index obj_origin;
863
+
trace
864
+
in
865
+
t.trace <- trace;
866
+
if acquire then (
867
+
(* acquire immediate children, and so on recursively *)
868
+
match t.desc with
869
+
| Map (x, _) -> sub_acquire self x
870
+
| Map2 (x, y, _) ->
871
+
sub_acquire self x;
872
+
sub_acquire self y
873
+
| Pair (x, y) ->
874
+
sub_acquire self x;
875
+
sub_acquire self y
876
+
| App (x, y) ->
877
+
sub_acquire self x;
878
+
sub_acquire self y
879
+
| Fix {doc; wrt} ->
880
+
sub_acquire self doc;
881
+
sub_acquire self wrt
882
+
| Join { child; intermediate } ->
883
+
sub_acquire self child;
884
+
begin match intermediate with
885
+
| None -> ()
886
+
| Some _ ->
887
+
assert false (* this can't initialized already, first-time acquire *)
888
+
end
889
+
| Var _ -> ()
890
+
| Prim _ -> ()
891
+
)
892
+
893
+
(* make sure that [origin] is in [self.trace], passed as last arg. *)
894
+
let activate_tracing self origin = function
895
+
| Tn tn ->
896
+
let idx = get_idx self origin in (* index of [self] in [origin.trace_idx] *)
897
+
let active = tn.active in
898
+
(* [idx < active] means [self] is already traced by [origin].
899
+
We only have to add [self] to the entries if [idx >= active]. *)
900
+
if idx >= active then (
901
+
tn.active <- active + 1;
902
+
);
903
+
if idx > active then (
904
+
(* swap with last entry in [tn.entries] *)
905
+
let old = tn.entries.(active) in
906
+
tn.entries.(idx) <- old;
907
+
tn.entries.(active) <- obj_t origin;
908
+
mov_idx self active idx old;
909
+
mov_idx self idx active origin
910
+
)
364
911
| _ -> ()
365
-
)
366
-
| _ -> assert false
912
+
913
+
let sub_is_damaged = function
914
+
| Root _ -> assert false
915
+
| Pure _ -> false
916
+
| Operator {value; _} ->
917
+
match value with
918
+
| Eval_none | Eval_invalid_next -> true
919
+
| Eval_some _ -> false
920
+
| Eval_progress -> assert false
921
+
922
+
(* [sub_sample origin self] computes a value for [self].
923
+
924
+
[sub_sample] raise if any user-provided computation raises.
925
+
Graph will be left in a coherent state but exception will be propagated
926
+
to the observer. *)
927
+
let rec sub_sample queue =
928
+
let rec aux : type a b . a t_ -> b t_ -> b = fun origin ->
929
+
function
930
+
| Root _ -> assert false
931
+
| Pure x -> x
932
+
| Operator t as self ->
933
+
(* lock the mutex, examine cached value *)
367
934
368
-
let peek_stable = function
369
-
| Operator ({desc = Var v; _}) -> v.binding
370
-
| _ -> assert false
935
+
Mutex.lock t.mutex;
371
936
372
-
let peek = function
373
-
| Operator ({desc = Var v; _}) -> v.nextVal |>Option.value ~default: v.binding
374
-
| _ -> assert false
937
+
match t.value with
938
+
| Eval_some value ->
939
+
Mutex.unlock t.mutex;
940
+
activate_tracing self origin t.trace;
941
+
value
942
+
| Eval_none ->
943
+
t.value <- Eval_progress;
944
+
Mutex.unlock t.mutex;
945
+
946
+
(* compute value without holding the lock *)
375
947
376
-
let update f v = set v (f (peek v))
948
+
let result : b =
949
+
match t.desc with
950
+
| Map (x, f) -> f (aux self x)
951
+
| Map2 (x, y, f) -> f (aux self x) (aux self y)
952
+
| Pair (x, y) -> (aux self x, aux self y)
953
+
| App (f, x) -> (aux self f) (aux self x)
954
+
| Fix { doc; wrt } ->
955
+
let _ = aux self wrt in
956
+
let result = aux self doc in
957
+
if sub_is_damaged wrt then aux origin self
958
+
else (
959
+
if sub_is_damaged doc then do_invalidate Fragile self;
960
+
result)
961
+
| Join x ->
377
962
378
-
let may_update f v =
379
-
match f (peek v) with
380
-
| None -> ()
381
-
| Some x -> set v x
963
+
let intermediate =
964
+
(* We haven't touched any state yet,
965
+
it is safe for [aux] to raise *)
966
+
aux self x.child
967
+
in
968
+
begin
969
+
match x.intermediate with
970
+
| None ->
971
+
x.intermediate <- Some intermediate;
382
972
383
-
(* Primitives *)
384
-
type 'a prim = 'a t
385
-
let prim ~acquire ~release =
386
-
inj (operator (Prim { acquire; release }))
387
-
let get_prim x = x
973
+
sub_acquire self intermediate
974
+
| Some x' when x' != intermediate ->
975
+
queue :=
976
+
Release_more
977
+
{ origin = self; element = x'; next = !queue };
978
+
x.intermediate <- Some intermediate;
388
979
389
-
let invalidate x = match prj x with
390
-
| Operator {desc = Prim p; value; _} as t ->
391
-
(* the value is invalidated, be sure to invalidate all parents as well *)
392
-
begin match value with
393
-
| Eval_none | Eval_invalid_next -> ()
394
-
| Eval_progress -> do_invalidate Fragile t;
395
-
| Eval_some v ->
396
-
do_invalidate Strong t;
397
-
p.release x v
398
-
end
399
-
| _ -> assert false
980
+
sub_acquire self intermediate
981
+
| Some _ -> ()
982
+
end;
983
+
(*print mermaid*)
400
984
401
-
(* Fix point *)
985
+
let mermaid=to_mermaid_trace_idx ~max_nodes:200 ( intermediate) in
986
+
(* let mermaid_2=to_mermaid ~max_nodes:200 ( intermediate) in *)
987
+
let mermaid_3=to_mermaid_trace_idx ~max_nodes:200 ( intermediate) in
402
988
403
-
let fix doc ~wrt = match prj wrt with
404
-
| Root _ -> assert false
405
-
| Pure _ -> doc
406
-
| Operator _ as wrt -> inj (operator (Fix {doc = prj doc; wrt}))
407
989
408
-
type release_list =
409
-
| Release_done
410
-
| Release_more :
411
-
{ origin : 'a t_; element : 'b t_; next : release_list } -> release_list
412
990
413
-
type release_queue = release_list ref
414
-
let make_release_queue () = ref Release_done
991
+
aux self intermediate
992
+
| Var x ->
415
993
416
-
type release_failure = exn * Printexc.raw_backtrace
994
+
x.binding
995
+
| Prim t -> t.acquire (inj self)
996
+
in
997
+
998
+
(* lock again and finalize *)
417
999
418
-
(* [sub_release [] origin self] is called when [origin] is released,
419
-
where [origin] is reachable from [self]'s trace.
420
-
We're going to remove [origin] from that trace as [origin] is now dead.
1000
+
Mutex.lock t.mutex;
1001
+
begin
1002
+
match t.value with
1003
+
| Eval_progress -> t.value <- Eval_some result
1004
+
| Eval_invalid_next -> t.value <- Eval_none
1005
+
| Eval_none | Eval_some _ -> ()
1006
+
end;
1007
+
Mutex.unlock t.mutex;
421
1008
422
-
[sub_release] cannot raise.
423
-
If a primitive raises, the exception is caught and a warning is emitted. *)
424
-
let rec sub_release
425
-
: type a b . release_failure list -> a t_ -> b t_ -> release_failure list
426
-
= fun failures origin -> function
427
-
| Root _ -> assert false
428
-
| Pure _ -> failures
429
-
| Operator t as self ->
430
-
(* compute [t.trace \ {origin}] *)
431
-
let trace = match t.trace with
432
-
| T0 -> assert false
433
-
| T1 x -> assert (t_equal x origin); T0
434
-
| T2 (x, y) ->
435
-
if t_equal x origin then T1 y
436
-
else if t_equal y origin then T1 x
437
-
else assert false
438
-
| T3 (x, y, z) ->
439
-
if t_equal x origin then T2 (y, z)
440
-
else if t_equal y origin then T2 (x, z)
441
-
else if t_equal z origin then T2 (x, y)
442
-
else assert false
443
-
| T4 (x, y, z, w) ->
444
-
if t_equal x origin then T3 (y, z, w)
445
-
else if t_equal y origin then T3 (x, z, w)
446
-
else if t_equal z origin then T3 (x, y, w)
447
-
else if t_equal w origin then T3 (x, y, z)
448
-
else assert false
449
-
| Tn tn as trace ->
450
-
let revidx = rem_idx self origin in
451
-
assert (t_equal tn.entries.(revidx) origin);
452
-
let count = tn.count - 1 in
453
-
tn.count <- count;
454
-
if revidx < count then (
455
-
let obj = tn.entries.(count) in
456
-
tn.entries.(revidx) <- obj;
457
-
tn.entries.(count) <- dummy;
458
-
mov_idx self count revidx obj
459
-
) else
460
-
tn.entries.(revidx) <- dummy;
461
-
if tn.active > count then tn.active <- count;
462
-
if count = 4 then (
463
-
(* downgrade to [T4] to save space *)
464
-
let a = tn.entries.(0) and b = tn.entries.(1) in
465
-
let c = tn.entries.(2) and d = tn.entries.(3) in
466
-
ignore (rem_idx self a : int);
467
-
ignore (rem_idx self b : int);
468
-
ignore (rem_idx self c : int);
469
-
ignore (rem_idx self d : int);
470
-
T4 (a, b, c, d)
471
-
) else (
472
-
let len = Array.length tn.entries in
473
-
if count <= len lsr 2 then
474
-
Tn { active = tn.active; count = tn.count;
475
-
entries = Array.sub tn.entries 0 (len lsr 1) }
476
-
else
477
-
trace
478
-
)
479
-
in
480
-
t.trace <- trace;
481
-
match trace with
482
-
| T0 ->
483
-
(* [self] is not active anymore, since it's not reachable
484
-
from any root. We can release its cached value and
485
-
recursively release its subtree. *)
486
-
let value = t.value in
487
-
t.value <- Eval_progress;
488
-
begin match t.desc with
489
-
| Map (x, _) -> sub_release failures self x
490
-
| Map2 (x, y, _) ->
491
-
sub_release (sub_release failures self x) self y
492
-
| Pair (x, y) ->
493
-
sub_release (sub_release failures self x) self y
494
-
| App (x, y) ->
495
-
sub_release (sub_release failures self x) self y
496
-
| Join ({ child; intermediate } as t) ->
497
-
let failures = sub_release failures self child in
498
-
begin match intermediate with
499
-
| None -> failures
500
-
| Some child' ->
501
-
t.intermediate <- None;
502
-
sub_release failures self child'
503
-
end
504
-
| Var _ -> failures
505
-
| Fix {doc; wrt} ->
506
-
sub_release (sub_release failures self wrt) self doc
507
-
| Prim t ->
508
-
begin match value with
509
-
| Eval_none | Eval_invalid_next | Eval_progress -> failures
510
-
| Eval_some x ->
511
-
begin match t.release (inj self) x with
512
-
| () -> failures
513
-
| exception exn ->
514
-
let bt = Printexc.get_raw_backtrace () in
515
-
(exn, bt) :: failures
516
-
end
517
-
end
518
-
end
519
-
| _ -> failures
1009
+
520
1010
521
-
(* [sub_acquire] cannot raise *)
522
-
let rec sub_acquire : type a b . a t_ -> b t_ -> unit = fun origin ->
523
-
function
524
-
| Root _ -> assert false
525
-
| Pure _ -> ()
526
-
| Operator t as self ->
527
-
(*lock the mutex, because we are making changes within this node *)
528
-
Mutex.protect t.mutex @@ fun _->
529
-
(* [acquire] is true if this is the first time this operator
530
-
is used, in which case we need to acquire its children *)
531
-
let acquire = match t.trace with T0 -> true | _ -> false in
532
-
let trace = match t.trace with
533
-
| T0 -> T1 origin
534
-
| T1 x -> T2 (origin, x)
535
-
| T2 (x, y) -> T3 (origin, x, y)
536
-
| T3 (x, y, z) -> T4 (origin, x, y, z)
537
-
| T4 (x, y, z, w) ->
538
-
let obj_origin = obj_t origin in
539
-
let entries =
540
-
[| obj_t x; obj_t y; obj_t z; obj_t w; obj_origin; dummy; dummy; dummy |]
541
-
in
542
-
for i = 0 to 4 do add_idx self i entries.(i) done;
543
-
Tn { active = 5; count = 5; entries }
544
-
| Tn tn as trace ->
545
-
let index = tn.count in
546
-
let entries, trace =
547
-
(* possibly resize array [entries] *)
548
-
if index < Array.length tn.entries then (
549
-
tn.count <- tn.count + 1;
550
-
(tn.entries, trace)
551
-
) else (
552
-
let entries = Array.make (index * 2) dummy in
553
-
Array.blit tn.entries 0 entries 0 index;
554
-
(entries, Tn { active = tn.active; count = index + 1; entries })
555
-
)
556
-
in
557
-
let obj_origin = obj_t origin in
558
-
entries.(index) <- obj_origin;
559
-
add_idx self index obj_origin;
560
-
trace
561
-
in
562
-
t.trace <- trace;
563
-
if acquire then (
564
-
(* acquire immediate children, and so on recursively *)
565
-
match t.desc with
566
-
| Map (x, _) -> sub_acquire self x
567
-
| Map2 (x, y, _) ->
568
-
sub_acquire self x;
569
-
sub_acquire self y
570
-
| Pair (x, y) ->
571
-
sub_acquire self x;
572
-
sub_acquire self y
573
-
| App (x, y) ->
574
-
sub_acquire self x;
575
-
sub_acquire self y
576
-
| Fix {doc; wrt} ->
577
-
sub_acquire self doc;
578
-
sub_acquire self wrt
579
-
| Join { child; intermediate } ->
580
-
sub_acquire self child;
581
-
begin match intermediate with
582
-
| None -> ()
583
-
| Some _ ->
584
-
assert false (* this can't initialized already, first-time acquire *)
585
-
end
586
-
| Var _ -> ()
587
-
| Prim _ -> ()
588
-
)
1011
+
(* Re-evaluate if the node was invalidated during computation *)
1012
+
if t.value == Eval_none then (
589
1013
590
-
(* make sure that [origin] is in [self.trace], passed as last arg. *)
591
-
let activate_tracing self origin = function
592
-
| Tn tn ->
593
-
let idx = get_idx self origin in (* index of [self] in [origin.trace_idx] *)
594
-
let active = tn.active in
595
-
(* [idx < active] means [self] is already traced by [origin].
596
-
We only have to add [self] to the entries if [idx >= active]. *)
597
-
if idx >= active then (
598
-
tn.active <- active + 1;
599
-
);
600
-
if idx > active then (
601
-
(* swap with last entry in [tn.entries] *)
602
-
let old = tn.entries.(active) in
603
-
tn.entries.(idx) <- old;
604
-
tn.entries.(active) <- obj_t origin;
605
-
mov_idx self active idx old;
606
-
mov_idx self idx active origin
607
-
)
608
-
| _ -> ()
1014
+
aux origin self
1015
+
) else (
609
1016
610
-
let sub_is_damaged = function
611
-
| Root _ -> assert false
612
-
| Pure _ -> false
613
-
| Operator {value; _} ->
614
-
match value with
615
-
| Eval_none | Eval_invalid_next -> true
616
-
| Eval_some _ -> false
617
-
| Eval_progress -> assert false
1017
+
(* [self] just became active, so it may invalidate [origin] in case its
1018
+
value changes because of [t.desc], like if it's a variable and gets
1019
+
mutated, or if it's a primitive that gets invalidated.
1020
+
We need to put [origin] into [self.trace] in case it isn't there yet. *)
1021
+
activate_tracing self origin t.trace;
1022
+
result)
1023
+
| Eval_progress | Eval_invalid_next ->
1024
+
Mutex.unlock t.mutex;
618
1025
619
-
(* [sub_sample origin self] computes a value for [self].
1026
+
(* spin and retry *)
1027
+
let rec spin () =
1028
+
match t.value with
1029
+
| Eval_progress | Eval_invalid_next ->
1030
+
Domain.cpu_relax ();
1031
+
spin ()
1032
+
| Eval_none | Eval_some _ -> ()
1033
+
in
1034
+
spin ();
620
1035
621
-
[sub_sample] raise if any user-provided computation raises.
622
-
Graph will be left in a coherent state but exception will be propagated
623
-
to the observer. *)
624
-
let sub_sample queue =
625
-
let rec aux : type a b . a t_ -> b t_ -> b = fun origin ->
626
-
function
627
-
| Root _ -> assert false
628
-
| Pure x -> x
629
-
| Operator t as self ->
630
-
(* lock the mutex, examine cached value *)
631
-
Mutex.lock t.mutex;
632
-
match t.value with
633
-
| Eval_some value ->
634
-
Mutex.unlock t.mutex;
635
-
activate_tracing self origin t.trace;
636
-
value
637
-
| Eval_none ->
638
-
t.value <- Eval_progress;
639
-
Mutex.unlock t.mutex;
1036
+
aux origin self
1037
+
in
1038
+
aux
1039
+
1040
+
type 'a root = 'a t
1041
+
1042
+
let observe ?(on_invalidate = ignore) child : _ root =
1043
+
let root =
1044
+
Root
1045
+
{ child = prj child
1046
+
; value = Eval_none
1047
+
; on_invalidate
1048
+
; trace_idx = I0
1049
+
; acquired = false
1050
+
; mutex= Mutex.create()
1051
+
}
1052
+
in
1053
+
inj root
1054
+
1055
+
exception Release_failure of exn option * release_failure list
1056
+
1057
+
let raw_flush_release_queue queue =
1058
+
let rec aux failures = function
1059
+
| Release_done -> failures
1060
+
| Release_more t ->
640
1061
641
-
(* compute value without holding the lock *)
642
-
let result : b =
643
-
match t.desc with
644
-
| Map (x, f) -> f (aux self x)
645
-
| Map2 (x, y, f) -> f (aux self x) (aux self y)
646
-
| Pair (x, y) -> (aux self x, aux self y)
647
-
| App (f, x) -> (aux self f) (aux self x)
648
-
| Fix { doc; wrt } ->
649
-
let _ = aux self wrt in
650
-
let result = aux self doc in
651
-
if sub_is_damaged wrt then aux origin self
652
-
else (
653
-
if sub_is_damaged doc then do_invalidate Fragile self;
654
-
result)
655
-
| Join x ->
656
-
let intermediate =
657
-
(* We haven't touched any state yet,
658
-
it is safe for [aux] to raise *)
659
-
aux self x.child
660
-
in
661
-
begin
662
-
match x.intermediate with
663
-
| None ->
664
-
x.intermediate <- Some intermediate;
665
-
sub_acquire self intermediate
666
-
| Some x' when x' != intermediate ->
667
-
queue :=
668
-
Release_more
669
-
{ origin = self; element = x'; next = !queue };
670
-
x.intermediate <- Some intermediate;
671
-
sub_acquire self intermediate
672
-
| Some _ -> ()
673
-
end;
674
-
aux self intermediate
675
-
| Var x -> x.binding
676
-
| Prim t -> t.acquire (inj self)
677
-
in
1062
+
let failures = sub_release failures t.origin t.element in
1063
+
let return = aux failures t.next in
678
1064
679
-
(* lock again and finalize *)
680
-
Mutex.lock t.mutex;
681
-
begin
682
-
match t.value with
683
-
| Eval_progress -> t.value <- Eval_some result
684
-
| Eval_invalid_next -> t.value <- Eval_none
685
-
| Eval_none | Eval_some _ -> ()
686
-
end;
687
-
Mutex.unlock t.mutex;
1065
+
return
1066
+
in
1067
+
aux [] queue
1068
+
1069
+
let flush_release_queue queue =
1070
+
let queue' = !queue in
1071
+
queue := Release_done;
1072
+
raw_flush_release_queue queue'
1073
+
1074
+
let sample queue x = match prj x with
1075
+
| Pure _ | Operator _ -> assert false
1076
+
| Root t as self ->
688
1077
689
-
(* Re-evaluate if the node was invalidated during computation *)
690
-
if t.value == Eval_none then aux origin self
691
-
else (
692
-
(* [self] just became active, so it may invalidate [origin] in case its
693
-
value changes because of [t.desc], like if it's a variable and gets
694
-
mutated, or if it's a primitive that gets invalidated.
695
-
We need to put [origin] into [self.trace] in case it isn't there yet. *)
696
-
activate_tracing self origin t.trace;
697
-
result)
698
-
| Eval_progress | Eval_invalid_next ->
699
-
Mutex.unlock t.mutex;
700
-
(* spin and retry *)
701
-
let rec spin () =
702
-
match t.value with
703
-
| Eval_progress | Eval_invalid_next ->
704
-
Domain.cpu_relax ();
705
-
spin ()
706
-
| Eval_none | Eval_some _ -> ()
707
-
in
708
-
spin ();
709
-
aux origin self
710
-
in
711
-
aux
1078
+
(* debug log the whole tree *)
712
1079
713
-
type 'a root = 'a t
1080
+
let mermaid=to_mermaid_trace_idx ~max_nodes:200 ( t.child) in
1081
+
(* m "sample: graph: \n %s" mermaid); *)
1082
+
714
1083
715
-
let observe ?(on_invalidate = ignore) child : _ root =
716
-
let root =
717
-
Root
718
-
{ child = prj child
719
-
; value = Eval_none
720
-
; on_invalidate
721
-
; trace_idx = I0
722
-
; acquired = false
723
-
; mutex= Mutex.create()
724
-
}
725
-
in
726
-
inj root
727
1084
728
-
exception Release_failure of exn option * release_failure list
1085
+
(*lock the root mutex while sampling*)
729
1086
730
-
let raw_flush_release_queue queue =
731
-
let rec aux failures = function
732
-
| Release_done -> failures
733
-
| Release_more t ->
734
-
let failures = sub_release failures t.origin t.element in
735
-
aux failures t.next
736
-
in
737
-
aux [] queue
1087
+
let a=Mutex.protect t.mutex @@ fun _->
1088
+
match t.value with
1089
+
| Eval_some value -> value
1090
+
| _ ->
1091
+
(
1092
+
(* no cached value, compute it now *)
1093
+
if not t.acquired then (
738
1094
739
-
let flush_release_queue queue =
740
-
let queue' = !queue in
741
-
queue := Release_done;
742
-
raw_flush_release_queue queue'
1095
+
t.acquired <- true;
1096
+
let res = sub_acquire self t.child in
743
1097
744
-
let sample queue x = match prj x with
745
-
| Pure _ | Operator _ -> assert false
746
-
| Root t as self ->
747
-
(*lock the root mutex while sampling*)
748
-
Mutex.protect t.mutex @@ fun _->
749
-
match t.value with
750
-
| Eval_some value -> value
751
-
| _ ->
752
-
(
753
-
(* no cached value, compute it now *)
754
-
if not t.acquired then (
755
-
t.acquired <- true;
756
-
sub_acquire self t.child;
757
-
);
758
-
t.value <- Eval_progress;
759
-
let value = sub_sample queue self t.child in
760
-
begin match t.value with
761
-
| Eval_progress -> t.value <- Eval_some value; (* cache value *)
762
-
| Eval_none | Eval_some _ | Eval_invalid_next -> ()
763
-
end;
764
-
value
765
-
)
1098
+
res
1099
+
);
1100
+
1101
+
t.value <- Eval_progress;
1102
+
let value = sub_sample queue self t.child in
1103
+
begin match t.value with
1104
+
| Eval_progress ->
766
1105
767
-
let is_damaged x =
768
-
match prj x with
769
-
| Pure _ | Operator _ -> assert false
770
-
| Root {value;_}->
771
-
(* NOTE: I don't think i need a mutex here*)
772
-
(match value with
773
-
| Eval_some _ -> false
774
-
| Eval_none | Eval_progress | Eval_invalid_next -> true
1106
+
t.value <- Eval_some value; (* cache value *)
1107
+
| Eval_none | Eval_some _ | Eval_invalid_next -> ()
1108
+
end;
1109
+
value
775
1110
)
1111
+
in
776
1112
777
-
let release queue x = match prj x with
778
-
| Pure _ | Operator _ -> assert false
779
-
| Root t as self ->
780
-
Mutex.protect t.mutex @@ fun _->
781
-
if t.acquired then (
782
-
(* release subtree, remove cached value *)
783
-
t.value <- Eval_none;
784
-
t.acquired <- false;
785
-
queue := Release_more { origin = self; element = t.child; next = !queue }
786
-
)
787
-
788
-
let set_on_invalidate x f =
789
-
match prj x with
790
-
| Pure _ | Operator _ -> assert false
791
-
| Root t ->
792
-
t.on_invalidate <- f
793
-
794
-
let flush_or_fail main_exn queue =
795
-
match flush_release_queue queue with
796
-
| [] -> ()
797
-
| failures -> raise (Release_failure (main_exn, failures))
798
-
799
-
let quick_sample root =
800
-
let queue = ref Release_done in
801
-
match sample queue root with
802
-
| result -> flush_or_fail None queue; result
803
-
| exception exn -> flush_or_fail (Some exn) queue; raise exn
804
-
805
-
let quick_release root =
806
-
let queue = ref Release_done in
807
-
release queue root;
808
-
flush_or_fail None queue
809
-
810
-
module Infix = struct
811
-
let (>>=) x f = bind x ~f
812
-
let (>|=) x f = map x ~f
813
-
let (<*>) = app
814
-
end
815
-
816
-
(*$R
817
-
let x = var 0 in
818
-
let y = map ~f:succ (get x) in
819
-
let o_y = Lwd.observe y in
820
-
assert_equal 1 (quick_sample o_y);
821
-
set x 10;
822
-
assert_equal 11 (quick_sample o_y);
823
-
*)
824
-
end
1113
+
a
1114
+
1115
+
let is_damaged x =
1116
+
match prj x with
1117
+
| Pure _ | Operator _ -> assert false
1118
+
| Root {value;_}->
1119
+
(* NOTE: I don't think i need a mutex here*)
1120
+
(match value with
1121
+
| Eval_some _ -> false
1122
+
| Eval_none | Eval_progress | Eval_invalid_next -> true
1123
+
)
1124
+
1125
+
let release queue x = match prj x with
1126
+
| Pure _ | Operator _ -> assert false
1127
+
| Root t as self ->
1128
+
Mutex.protect t.mutex @@ fun _->
1129
+
if t.acquired then (
1130
+
(* release subtree, remove cached value *)
1131
+
t.value <- Eval_none;
1132
+
t.acquired <- false;
1133
+
queue := Release_more { origin = self; element = t.child; next = !queue }
1134
+
)
1135
+
1136
+
let set_on_invalidate x f =
1137
+
match prj x with
1138
+
| Pure _ | Operator _ -> assert false
1139
+
| Root t ->
1140
+
t.on_invalidate <- f
1141
+
1142
+
let flush_or_fail main_exn queue =
1143
+
match flush_release_queue queue with
1144
+
| [] -> ()
1145
+
| failures -> raise (Release_failure (main_exn, failures))
1146
+
1147
+
let quick_sample root =
1148
+
let queue = ref Release_done in
1149
+
match sample queue root with
1150
+
| result -> flush_or_fail None queue; result
1151
+
| exception exn -> flush_or_fail (Some exn) queue; raise exn
1152
+
1153
+
let quick_release root =
1154
+
let queue = ref Release_done in
1155
+
release queue root;
1156
+
flush_or_fail None queue
1157
+
1158
+
module Infix = struct
1159
+
let (>>=) x f = bind x ~f
1160
+
let (>|=) x f = map x ~f
1161
+
let (<*>) = app
1162
+
end
1163
+
1164
+
(*$R
1165
+
let x = var 0 in
1166
+
let y = map ~f:succ (get x) in
1167
+
let o_y = Lwd.observe y in
1168
+
assert_equal 1 (quick_sample o_y);
1169
+
set x 10;
1170
+
assert_equal 11 (quick_sample o_y);
1171
+
*)
1172
+
end
+6
-1
forks/lwd/lib/lwd/mutex_backend.ml
+6
-1
forks/lwd/lib/lwd/mutex_backend.ml
···
1
1
(** Backend selection for mutex implementations. *)
2
2
3
3
module type MUTEX = sig
4
-
include module type of Mutex
4
+
type t
5
+
val create : unit -> t
6
+
val lock : t -> unit
7
+
val unlock : t -> unit
8
+
val try_lock : t -> bool
9
+
val protect : t -> (unit -> 'a) -> 'a
5
10
val lock_all : t list -> bool
6
11
end
7
12
+7
-4
forks/lwd/lib/lwd_picos/dune
+7
-4
forks/lwd/lib/lwd_picos/dune
···
19
19
20
20
;; Optional Picos backend - only compiled if picos is available
21
21
(library
22
+
; (package lwd)
22
23
(name lwd_picos)
23
-
(modules lwd mutex_picos)
24
-
(libraries picos picos_stdlwd)
25
-
(optional)
26
-
(wrapped false))
24
+
(public_name lwd_picos)
25
+
(libraries picos picos_std.structured picos_std.sync)
26
+
27
+
(implements lwd)
28
+
29
+
)
27
30
28
31
; (executable
29
32
; (name pp)
+7
-1
forks/lwd/lib/lwd_picos/lwd.ml
+7
-1
forks/lwd/lib/lwd_picos/lwd.ml
-1
forks/lwd/lib/lwd_picos/lwd.mli
-1
forks/lwd/lib/lwd_picos/lwd.mli
···
1
-
include module type of Lwd.Lwd
+7
-64
forks/lwd/lib/lwd_picos/mutex_picos.ml
+7
-64
forks/lwd/lib/lwd_picos/mutex_picos.ml
···
8
8
Since Picos doesn't have a direct mutex equivalent, we'll implement
9
9
one using Picos's basic synchronization primitives. *)
10
10
11
-
type t = {
12
-
mutable locked : bool;
13
-
mutable owner : Picos.Fiber.t option;
14
-
waiters : Picos.Trigger.t list ref;
15
-
}
11
+
include Picos_std_sync.Mutex
16
12
17
-
let create () = {
18
-
locked = false;
19
-
owner = None;
20
-
waiters = ref [];
21
-
}
13
+
let create () = create ()
22
14
23
-
let lock mutex =
24
-
let rec try_acquire () =
25
-
if not mutex.locked then begin
26
-
mutex.locked <- true;
27
-
mutex.owner <- Some (Picos.Fiber.current ());
28
-
end else begin
29
-
(* Create a trigger to wait for the mutex to be released *)
30
-
let trigger = Picos.Trigger.create () in
31
-
mutex.waiters := trigger :: !(mutex.waiters);
32
-
match Picos.await trigger with
33
-
| None -> (* Cancelled *) raise (Sys_error "Mutex lock cancelled")
34
-
| Some (exn, _) -> raise exn
35
-
end
36
-
in
37
-
try_acquire ()
15
+
let lock mut = lock mut
38
16
39
-
let try_lock mutex =
40
-
if not mutex.locked then begin
41
-
mutex.locked <- true;
42
-
mutex.owner <- Some (Picos.Fiber.current ());
43
-
true
44
-
end else
45
-
false
46
-
47
-
let unlock mutex =
48
-
if not mutex.locked then
49
-
raise (Sys_error "Mutex is not locked")
50
-
else
51
-
let current_fiber = Picos.Fiber.current () in
52
-
match mutex.owner with
53
-
| None -> raise (Sys_error "Mutex has no owner")
54
-
| Some owner ->
55
-
if not (Picos.Fiber.equal current_fiber owner) then
56
-
raise (Sys_error "Mutex was locked by another fiber")
57
-
else begin
58
-
mutex.locked <- false;
59
-
mutex.owner <- None;
60
-
(* Wake up one waiter if any *)
61
-
match !(mutex.waiters) with
62
-
| [] -> ()
63
-
| trigger :: rest ->
64
-
mutex.waiters := rest;
65
-
Picos.Trigger.signal trigger ()
66
-
end
67
-
68
-
let protect mutex f =
69
-
lock mutex;
70
-
try
71
-
let result = f () in
72
-
unlock mutex;
73
-
result
74
-
with exn ->
75
-
unlock mutex;
76
-
raise exn
17
+
let unlock mut = unlock mut
18
+
let try_lock mut = try_lock mut
19
+
let protect mut f = protect mut f
77
20
78
21
let lock_all mutexes =
79
22
let rec try_lock_all acc = function
···
89
32
false
90
33
end
91
34
in
92
-
try_lock_all [] mutexes
35
+
try_lock_all [] mutexes
+1
-1
forks/lwd/lib/lwd_picos/mutex_picos.mli
+1
-1
forks/lwd/lib/lwd_picos/mutex_picos.mli
+34
forks/lwd/lib/lwd_stdlib/dune
+34
forks/lwd/lib/lwd_stdlib/dune
···
1
+
; (library
2
+
; (name lwd)
3
+
; (public_name lwd)
4
+
; (modules lwd lwd_seq lwd_table lwd_infix lwd_utils mutex_config)
5
+
; (modules_without_implementation mutex)
6
+
; (libraries seq)
7
+
; (inline_tests
8
+
; (backend qtest.lib)
9
+
; (executable
10
+
; (flags
11
+
; (-w -33))))
12
+
; (wrapped false)
13
+
; (preprocess
14
+
; (per_module
15
+
; ((action
16
+
; (run %{dep:pp.exe} %{input-file}))
17
+
; lwd_infix
18
+
; lwd_seq))))
19
+
20
+
;; Optional Picos backend - only compiled if picos is available
21
+
(library
22
+
; (package lwd)
23
+
24
+
(name lwd_stdlib)
25
+
(public_name lwd_stdlib)
26
+
(libraries lwd )
27
+
28
+
(implements lwd)
29
+
30
+
)
31
+
32
+
; (executable
33
+
; (name pp)
34
+
; (modules pp))
+1
forks/lwd/lib/lwd_stdlib/lwd.ml
+1
forks/lwd/lib/lwd_stdlib/lwd.ml
···
1
+
include Lwd_impl.Make(Mutex_stdlib)
+19
forks/lwd/lib/lwd_stdlib/mutex_stdlib.ml
+19
forks/lwd/lib/lwd_stdlib/mutex_stdlib.ml
···
1
+
(** OCaml standard library implementation of the mutex interface. *)
2
+
3
+
include Mutex
4
+
5
+
let lock_all mutexes =
6
+
let rec try_lock_all acc = function
7
+
| [] ->
8
+
(* All mutexes acquired successfully *)
9
+
true
10
+
| mutex :: rest ->
11
+
if try_lock mutex then
12
+
try_lock_all (mutex :: acc) rest
13
+
else begin
14
+
(* Failed to acquire current mutex, release all previously acquired ones *)
15
+
List.iter unlock acc;
16
+
false
17
+
end
18
+
in
19
+
try_lock_all [] mutexes
+28
forks/lwd/lwd_stdlib.opam
+28
forks/lwd/lwd_stdlib.opam
···
1
+
# This file is generated by dune, edit dune-project instead
2
+
opam-version: "2.0"
3
+
synopsis: "Lightweight reactive documents with Stdlib backend"
4
+
maintainer: ["fred@tarides.com"]
5
+
authors: ["Frédéric Bour"]
6
+
license: "MIT"
7
+
homepage: "https://github.com/let-def/lwd"
8
+
bug-reports: "https://github.com/let-def/lwd/issues"
9
+
depends: [
10
+
"dune" {>= "3.5"}
11
+
"lwd"
12
+
"odoc" {with-doc}
13
+
]
14
+
build: [
15
+
["dune" "subst"] {dev}
16
+
[
17
+
"dune"
18
+
"build"
19
+
"-p"
20
+
name
21
+
"-j"
22
+
jobs
23
+
"@install"
24
+
"@runtest" {with-test}
25
+
"@doc" {with-doc}
26
+
]
27
+
]
28
+
dev-repo: "git+https://github.com/let-def/lwd.git"