/
incr_map.ml
1915 lines (1810 loc) · 71.5 KB
/
incr_map.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
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
open! Core
include Incr_map_intf
let no_instrumentation = { Instrumentation.f = (fun f -> f ()) }
(** This type lets us capture the kind of map function being performed, so we can with
one implementation perform map and filter-map operations.
Here, ['input_data] is the type of data in the input map, ['output_data] is the type
of data in the output map, and ['f_output] is the return type of the [~f] function
passed to the mapping function. *)
module Map_type = struct
type ('input_data, 'output_data, 'f_output) t =
| Map : ('input_data, 'output_data, 'output_data) t
| Filter_map : ('input_data, 'output_data, 'output_data option) t
(* The extra type variable 'a is to allow in future:
| Filter : ('output_data, 'output_data, bool) t *)
end
module Generic = struct
let with_old ~instrumentation i ~f =
let open Incremental.Let_syntax in
let old = ref None in
let%map a = i in
instrumentation.Instrumentation.f (fun () ->
let b = f ~old:!old a in
old := Some (a, b);
b)
;;
let cutoff ?(instrumentation = no_instrumentation) map ~cutoff =
let data_equal old_value new_value =
Incremental.Cutoff.should_cutoff cutoff ~old_value ~new_value
in
with_old ~instrumentation map ~f:(fun ~old cur ->
match old with
| None -> cur
| Some (_old_in, old) ->
Map.fold_symmetric_diff ~data_equal ~init:old old cur ~f:(fun acc (key, change) ->
match change with
| `Left _old -> Map.remove acc key
| `Right new_ -> Map.add_exn acc ~key ~data:new_
| `Unequal (_old, new_value) -> Map.set acc ~key ~data:new_value))
;;
let unordered_fold
~instrumentation
?(data_equal = phys_equal)
?update
?specialized_initial
?(finalize = Fn.id)
?(revert_to_init_when_empty = false)
map
~init
~add
~remove
=
let update =
let default ~key ~old_data ~new_data acc =
add ~key ~data:new_data (remove ~key ~data:old_data acc)
in
Option.value update ~default
in
with_old ~instrumentation map ~f:(fun ~old new_in ->
let acc =
match old with
| None ->
(match specialized_initial with
| None -> Map.fold ~init ~f:add new_in
| Some initial -> initial ~init new_in)
| Some (old_in, old_out) ->
if revert_to_init_when_empty && Map.length new_in = 0
then init
else
Map.fold_symmetric_diff
~init:old_out
old_in
new_in
~data_equal
~f:(fun acc (key, change) ->
match change with
| `Left old -> remove ~key ~data:old acc
| `Right new_ -> add ~key ~data:new_ acc
| `Unequal (old, new_) -> update ~key ~old_data:old ~new_data:new_ acc)
in
finalize acc)
;;
let unordered_fold_nested_maps
~instrumentation
?(data_equal = phys_equal)
?revert_to_init_when_empty
?update
incr_map
~init
~add
~remove
=
let update =
match update with
| Some update -> update
| None ->
fun ~outer_key ~inner_key ~old_data ~new_data acc ->
add
~outer_key
~inner_key
~data:new_data
(remove ~outer_key ~inner_key ~data:old_data acc)
in
unordered_fold
incr_map
~instrumentation
?revert_to_init_when_empty
~init
~update:(fun ~key:outer_key ~old_data:old_inner_map ~new_data:new_inner_map acc ->
(Map.fold_symmetric_diff old_inner_map new_inner_map ~data_equal)
~init:acc
~f:(fun acc (inner_key, diff) ->
match diff with
| `Left data_removed -> remove ~outer_key ~inner_key ~data:data_removed acc
| `Right data_added -> add ~outer_key ~inner_key ~data:data_added acc
| `Unequal (old_data, new_data) ->
update ~outer_key ~inner_key ~old_data ~new_data acc) [@nontail])
~add:(fun ~key:outer_key ~data:inner_map acc ->
Map.fold inner_map ~init:acc ~f:(fun ~key:inner_key ~data acc ->
add ~outer_key ~inner_key ~data acc))
~remove:(fun ~key:outer_key ~data:inner_map acc ->
Map.fold inner_map ~init:acc ~f:(fun ~key:inner_key ~data acc ->
remove ~outer_key ~inner_key ~data acc))
;;
let with_comparator' get_comparator x f =
Incremental.bind (Incremental.freeze (Incremental.map x ~f:get_comparator)) ~f
;;
(** Captures the comparator (which can't change anyway, since the type determines the
comparator) by freezing the corresponding map. Note that by first using Incremental.map to
get the comparator out of the map, we allow the initial map itself to be garbage
collected *)
let with_comparator map f = with_comparator' Map.comparator map f
let of_set ?(instrumentation = no_instrumentation) set =
with_comparator' Set.comparator set (fun comparator ->
let old_input = ref (Set.Using_comparator.empty ~comparator) in
let old_output = ref (Map.Using_comparator.empty ~comparator) in
Incremental.map set ~f:(fun new_input ->
instrumentation.Instrumentation.f (fun () ->
let new_output =
Sequence.fold
(Set.symmetric_diff !old_input new_input)
~init:!old_output
~f:(fun output -> function
| First k -> Map.remove output k
| Second k -> Map.add_exn output ~key:k ~data:())
in
old_input := new_input;
old_output := new_output;
new_output)))
;;
let generic_mapi
(type input_data output_data f_output state_witness)
(witness : (input_data, output_data, f_output) Map_type.t)
~instrumentation
?(data_equal = phys_equal)
(map : (('key, input_data, 'cmp) Map.t, state_witness) Incremental.t)
~(f : key:'key -> data:input_data -> f_output)
=
with_old ~instrumentation map ~f:(fun ~old input ->
match old, Map.length input with
| _, 0 | None, _ ->
(match witness with
| Map_type.Map -> (Map.mapi input ~f : ('key, output_data, 'cmp) Map.t)
| Map_type.Filter_map -> Map.filter_mapi input ~f)
| Some (old_input, old_output), _ ->
Map.fold_symmetric_diff
old_input
input
~data_equal
~init:old_output
~f:(fun output (key, change) ->
match change with
| `Left _ -> Map.remove output key
| `Right new_data | `Unequal (_, new_data) ->
let res = f ~key ~data:new_data in
(match witness with
| Map_type.Map -> Map.set output ~key ~data:res
| Map_type.Filter_map ->
(match res with
| None -> Map.remove output key
| Some output_data -> Map.set output ~key ~data:output_data))))
;;
let mapi ?(instrumentation = no_instrumentation) ?data_equal map ~f =
generic_mapi Map ~instrumentation ?data_equal map ~f
;;
let filter_mapi ?(instrumentation = no_instrumentation) ?data_equal map ~f =
generic_mapi Filter_map ~instrumentation ?data_equal map ~f
;;
let map ?instrumentation ?data_equal map ~f =
mapi ?instrumentation ?data_equal map ~f:(fun ~key:_ ~data -> f data)
;;
let filter_map ?instrumentation ?data_equal map ~f =
filter_mapi ?instrumentation ?data_equal map ~f:(fun ~key:_ ~data -> f data)
;;
let with_old2 ~instrumentation i1 i2 ~f =
let old = ref None in
Incremental.map2 i1 i2 ~f:(fun a1 a2 ->
instrumentation.Instrumentation.f (fun () ->
let b = f ~old:!old a1 a2 in
old := Some (a1, a2, b);
b))
;;
let unordered_fold_with_extra
?(instrumentation = no_instrumentation)
?(data_equal = phys_equal)
?(extra_equal = phys_equal)
?update
?specialized_initial
?(finalize = Fn.id)
?(revert_to_init_when_empty = false)
map
extra
~init
~add
~remove
~extra_changed
=
let update =
let default ~key ~old_data ~new_data acc extra =
add ~key ~data:new_data (remove ~key ~data:old_data acc extra) extra
in
Option.value update ~default
in
with_old2 ~instrumentation map extra ~f:(fun ~old new_in new_extra ->
let acc =
match old with
| None ->
(match specialized_initial with
| None ->
Map.fold new_in ~init ~f:(fun ~key ~data acc -> add ~key ~data acc new_extra)
| Some initial -> initial ~init new_in new_extra)
| Some (old_in, old_extra, old_out) ->
let acc =
if extra_equal old_extra new_extra
then old_out
else extra_changed ~old_extra ~new_extra ~input:old_in old_out
in
if revert_to_init_when_empty && Map.length new_in = 0
then init
else
Map.fold_symmetric_diff
~init:acc
old_in
new_in
~data_equal
~f:(fun acc (key, change) ->
match change with
| `Left old -> remove ~key ~data:old acc new_extra
| `Right new_ -> add ~key ~data:new_ acc new_extra
| `Unequal (old, new_) ->
update ~key ~old_data:old ~new_data:new_ acc new_extra)
in
finalize acc)
;;
let mapi_count
(type a cmp)
?(instrumentation = no_instrumentation)
?(data_equal = phys_equal)
input
~(comparator : (module Comparator.S with type t = a and type comparator_witness = cmp))
~f
=
let module M = (val comparator) in
let add new_key acc =
Map.update acc new_key ~f:(function
| None -> 1
| Some n -> n + 1)
in
let remove new_key acc =
Map.change acc new_key ~f:(function
| None -> None
| Some 1 -> None
| Some n -> Some (n - 1))
in
unordered_fold
~instrumentation
~data_equal
input
~init:(Map.empty (module M))
~add:(fun ~key ~data acc -> add (f ~key ~data) acc)
~remove:(fun ~key ~data acc -> remove (f ~key ~data) acc)
~update:(fun ~key ~old_data ~new_data acc ->
let prev_key = f ~key ~data:old_data in
let new_key = f ~key ~data:new_data in
if M.comparator.compare prev_key new_key = 0
then acc
else acc |> remove prev_key |> add new_key)
;;
let map_count ?instrumentation ?data_equal input ~comparator ~f =
mapi_count ?instrumentation ?data_equal input ~comparator ~f:(fun ~key:_ ~data ->
f data)
;;
let min_helper map =
match Map.min_elt map with
| None -> None
| Some (min, _) -> Some min
;;
let max_helper map =
match Map.max_elt map with
| None -> None
| Some (max, _) -> Some max
;;
let bounds_helper map =
match Map.min_elt map, Map.max_elt map with
| None, None -> None
| Some (min, _), Some (max, _) -> Some (min, max)
| _ -> assert false
;;
let mapi_min ?instrumentation ?data_equal input ~comparator ~f =
Incremental.map
~f:min_helper
(mapi_count ?instrumentation ?data_equal input ~comparator ~f)
;;
let mapi_max ?instrumentation ?data_equal input ~comparator ~f =
Incremental.map
~f:max_helper
(mapi_count ?instrumentation ?data_equal input ~comparator ~f)
;;
let mapi_bounds ?instrumentation ?data_equal input ~comparator ~f =
Incremental.map
~f:bounds_helper
(mapi_count ?instrumentation ?data_equal input ~comparator ~f)
;;
let map_min ?instrumentation ?data_equal input ~comparator ~f =
mapi_min ?instrumentation ?data_equal input ~comparator ~f:(fun ~key:_ ~data ->
f data)
;;
let map_max ?instrumentation ?data_equal input ~comparator ~f =
mapi_max ?instrumentation ?data_equal input ~comparator ~f:(fun ~key:_ ~data ->
f data)
;;
let min_value ?instrumentation ?data_equal input ~comparator =
map_min ?instrumentation ?data_equal input ~comparator ~f:Fn.id
;;
let max_value ?instrumentation ?data_equal input ~comparator =
map_max ?instrumentation ?data_equal input ~comparator ~f:Fn.id
;;
let map_bounds ?instrumentation ?data_equal input ~comparator ~f =
mapi_bounds ?instrumentation ?data_equal input ~comparator ~f:(fun ~key:_ ~data ->
f data)
;;
let value_bounds ?instrumentation ?data_equal input ~comparator =
map_bounds ?instrumentation ?data_equal input ~comparator ~f:Fn.id
;;
let merge_shared_impl
~old
~new_left_map
~new_right_map
~data_equal_left
~data_equal_right
~f
=
let comparator = Map.comparator new_left_map in
let old_left_map, old_right_map, old_output =
match old with
| None ->
let empty = Map.Using_comparator.empty ~comparator in
empty, empty, empty
| Some x -> x
in
let left_diff =
Map.symmetric_diff old_left_map new_left_map ~data_equal:data_equal_left
in
let right_diff =
Map.symmetric_diff old_right_map new_right_map ~data_equal:data_equal_right
in
(* We merge the two sides of the diffs together so we can make sure to handle each
key exactly once. This relies on symmetric diff giving sorted output. *)
Sequence.merge_with_duplicates
left_diff
right_diff
~compare:(fun (left_key, _) (right_key, _) -> comparator.compare left_key right_key)
|> Sequence.fold ~init:old_output ~f:(fun output diff_element ->
let key =
match diff_element with
| Left (key, _) | Right (key, _) -> key
| Both ((left_key, _), (_right_key, _)) ->
(* comparison functions can be expensive! *)
(* assert (comparator.compare left_key right_key = 0); *)
left_key
in
f ~old_output ~key ~output ~diff_element)
;;
let new_data_from_diff_element = function
| `Left _ -> None
| `Right x | `Unequal (_, x) -> Some x
;;
let merge
?(instrumentation = no_instrumentation)
?(data_equal_left = phys_equal)
?(data_equal_right = phys_equal)
left_map
right_map
~f
=
with_old2
left_map
right_map
~instrumentation
~f:(fun ~old new_left_map new_right_map ->
merge_shared_impl
~old
~new_left_map
~new_right_map
~data_equal_left
~data_equal_right
~f:(fun ~old_output:_ ~key ~output ~diff_element ->
(* These values represent whether there is data for the given key in the new
input in the left and right map. *)
let left_data_opt, right_data_opt =
match diff_element with
| Both ((_, left_diff), (_, right_diff)) ->
new_data_from_diff_element left_diff, new_data_from_diff_element right_diff
| Left (_, left_diff) ->
new_data_from_diff_element left_diff, Map.find new_right_map key
| Right (_, right_diff) ->
Map.find new_left_map key, new_data_from_diff_element right_diff
in
let output_data_opt =
match left_data_opt, right_data_opt with
| None, None -> None
| Some x, None -> f ~key (`Left x)
| None, Some y -> f ~key (`Right y)
| Some x, Some y -> f ~key (`Both (x, y))
in
match output_data_opt with
| None -> Map.remove output key
| Some data -> Map.set output ~key ~data))
;;
let merge_both_some
?(instrumentation = no_instrumentation)
?(data_equal_left = phys_equal)
?(data_equal_right = phys_equal)
?(out_equal = phys_equal)
left_map
right_map
~f
=
with_old2
left_map
right_map
~instrumentation
~f:(fun ~old new_left_map new_right_map ->
let comparator = Map.comparator new_left_map in
let empty = Map.Using_comparator.empty ~comparator in
match Map.length new_left_map, Map.length new_right_map with
(* Because we only care about keys that are in both maps, if either map is
empty, bail early. *)
| 0, _ | _, 0 -> empty
| _ ->
merge_shared_impl
~old
~new_left_map
~new_right_map
~data_equal_left
~data_equal_right
~f:(fun ~old_output ~key ~output ~diff_element ->
let left_and_right_data_opt =
let open Option.Let_syntax in
match diff_element with
| Both ((_, left_diff), (_, right_diff)) ->
let%bind left_data = new_data_from_diff_element left_diff in
let%map right_data = new_data_from_diff_element right_diff in
left_data, right_data
| Left (_, left_diff) ->
let%bind left_data = new_data_from_diff_element left_diff in
let%map right_data = Map.find new_right_map key in
left_data, right_data
| Right (_, right_diff) ->
(* This match arm binds [right_data] first because the map lookup
is slower than calling [new_data_from_diff_element]. *)
let%bind right_data = new_data_from_diff_element right_diff in
let%map left_data = Map.find new_left_map key in
left_data, right_data
in
(* look for the previously computed value to see if we actually need to
add or remove the key. *)
let prev_out = Map.find old_output key in
match left_and_right_data_opt with
| Some (x, y) ->
let data = f ~key x y in
(match prev_out with
(* if the new result is the same as the old, don't change the map *)
| Some prev_out when out_equal data prev_out -> output
| None | Some _ -> Map.set output ~key ~data)
| None ->
(match prev_out with
| None -> output
| Some _ -> Map.remove output key)))
;;
let generic_mapi_with_comparator'
(type input_data output_data f_output state_witness)
(witness : (input_data, output_data, f_output) Map_type.t)
~instrumentation
?cutoff
?(data_equal = phys_equal)
(lhs : (('key, input_data, 'cmp) Map.t, state_witness) Incremental.t)
~(comparator : ('key, 'cmp) Comparator.t)
~(f :
key:'key
-> data:(input_data, state_witness) Incremental.t
-> (f_output, state_witness) Incremental.t)
: (('key, output_data, 'cmp) Map.t, state_witness) Incremental.t
=
let module E = Incremental.Expert in
let incremental_state = Incremental.state lhs in
let empty_map = Map.Using_comparator.empty ~comparator in
let prev_map = ref empty_map in
let prev_nodes = ref empty_map in
let acc : ('key, output_data, 'cmp) Map.t ref = ref empty_map in
let result = E.Node.create incremental_state (fun () -> !acc) in
let (on_inner_change : key:'key -> f_output -> unit) =
match witness with
| Map_type.Map -> fun ~key data -> acc := Map.set !acc ~key ~data
| Map_type.Filter_map ->
fun ~key opt ->
let old = !acc in
acc
:= (match opt with
| None -> Map.remove old key
| Some data -> Map.set old ~key ~data)
in
let rec lhs_change =
lazy
(Incremental.map lhs ~f:(fun map ->
instrumentation.Instrumentation.f (fun () ->
let new_nodes =
Map.fold_symmetric_diff
~data_equal
!prev_map
map
~init:!prev_nodes
~f:(fun nodes (key, changed) ->
match changed with
| `Unequal _ ->
let node, _dep = Map.find_exn nodes key in
E.Node.make_stale node;
nodes
| `Left _ ->
let node, dep = Map.find_exn nodes key in
let nodes = Map.remove nodes key in
E.Node.remove_dependency result dep;
acc := Map.remove !acc key;
E.Node.invalidate node;
nodes
| `Right _ ->
let node =
E.Node.create incremental_state (fun () ->
Map.find_exn !prev_map key)
in
Option.iter cutoff ~f:(fun c ->
Incremental.set_cutoff (E.Node.watch node) c);
E.Node.add_dependency node (E.Dependency.create (force lhs_change));
let user_function_dep =
E.Dependency.create
(f ~key ~data:(E.Node.watch node))
~on_change:(on_inner_change ~key)
in
E.Node.add_dependency result user_function_dep;
Map.set nodes ~key ~data:(node, user_function_dep))
in
prev_nodes := new_nodes;
prev_map := map)))
in
E.Node.add_dependency result (E.Dependency.create (force lhs_change));
E.Node.watch result
;;
let filter_mapi' ?(instrumentation = no_instrumentation) ?cutoff ?data_equal map ~f =
with_comparator map (fun comparator ->
generic_mapi_with_comparator'
Map_type.Filter_map
~instrumentation
?cutoff
?data_equal
map
~f
~comparator)
;;
let mapi' ?(instrumentation = no_instrumentation) ?cutoff ?data_equal map ~f =
with_comparator map (fun comparator ->
generic_mapi_with_comparator'
Map_type.Map
~instrumentation
?cutoff
?data_equal
map
~f
~comparator)
;;
let map' ?instrumentation ?cutoff ?data_equal map ~f =
mapi' ?instrumentation ?cutoff ?data_equal map ~f:(fun ~key:_ ~data -> f data)
;;
let filter_map' ?instrumentation ?cutoff ?data_equal map ~f =
filter_mapi' ?instrumentation ?cutoff ?data_equal map ~f:(fun ~key:_ ~data -> f data)
;;
let merge' ?instrumentation ?cutoff ?data_equal_left ?data_equal_right map1 map2 ~f =
merge
?instrumentation
?data_equal_left
?data_equal_right
map1
map2
~f:(fun ~key:_ diff -> Some diff)
|> filter_mapi' ?instrumentation ?cutoff ~f:(fun ~key ~data:diff -> f ~key diff)
;;
let unzip_mapi_with_comparator
(type v v1 v2 state_witness)
~instrumentation
?(data_equal : v -> v -> bool = phys_equal)
?(left_result_equal : v1 -> v1 -> bool = phys_equal)
?(right_result_equal : v2 -> v2 -> bool = phys_equal)
(input : (('key, v, 'cmp) Map.t, state_witness) Incremental.t)
~(comparator : ('key, 'cmp) Comparator.t)
~(f : key:'key -> data:v -> v1 * v2)
: (('key, v1, 'cmp) Map.t, state_witness) Incremental.t
* (('key, v2, 'cmp) Map.t, state_witness) Incremental.t
=
let module E = Incremental.Expert in
let incremental_state = Incremental.state input in
let empty_map = Map.Using_comparator.empty ~comparator in
let left_acc : ('key, v1, 'cmp) Map.t ref = ref empty_map in
let left_result = E.Node.create incremental_state (fun () -> !left_acc) in
let right_acc : ('key, v2, 'cmp) Map.t ref = ref empty_map in
let right_result = E.Node.create incremental_state (fun () -> !right_acc) in
let prev_map = ref empty_map in
let input_change =
Incremental.map input ~f:(fun map ->
instrumentation.Instrumentation.f (fun () ->
let left, right =
match Map.is_empty !prev_map, Map.is_empty map with
| true, true | false, true -> empty_map, empty_map
| true, false ->
(* Mapping on a map is way faster than symmetric diffing and then
building the maps up piece by piece, so we do this whenever we
transition from "empty" to "something", which will almost always
happen on the first stabilization. *)
let left =
Map.mapi map ~f:(fun ~key ~data ->
let l, _ = f ~key ~data in
l)
in
let right =
Map.mapi map ~f:(fun ~key ~data ->
let _, r = f ~key ~data in
r)
in
left, right
| false, false ->
Map.fold_symmetric_diff
~data_equal
!prev_map
map
~init:(!left_acc, !right_acc)
~f:(fun (left, right) (key, changed) ->
match changed with
| `Unequal (prev, new_) ->
let prev_a, prev_b = f ~key ~data:prev in
let new_a, new_b = f ~key ~data:new_ in
let left =
if left_result_equal prev_a new_a
then left
else Map.set left ~key ~data:new_a
in
let right =
if right_result_equal prev_b new_b
then right
else Map.set right ~key ~data:new_b
in
left, right
| `Left _ -> Map.remove left key, Map.remove right key
| `Right element ->
let a, b = f ~key ~data:element in
Map.set left ~key ~data:a, Map.set right ~key ~data:b)
in
if not (phys_equal !left_acc left) then E.Node.make_stale left_result;
if not (phys_equal !right_acc right) then E.Node.make_stale right_result;
left_acc := left;
right_acc := right;
prev_map := map))
in
E.Node.add_dependency left_result (E.Dependency.create input_change);
E.Node.add_dependency right_result (E.Dependency.create input_change);
E.Node.watch left_result, E.Node.watch right_result
;;
let unzip_mapi
?(instrumentation = no_instrumentation)
?data_equal
?left_result_equal
?right_result_equal
input
~f
=
let pair =
with_comparator input (fun comparator ->
input
|> unzip_mapi_with_comparator
~instrumentation
?data_equal
?left_result_equal
?right_result_equal
~comparator
~f
|> Tuple2.uncurry Incremental.both)
in
Incremental.map ~f:fst pair, Incremental.map ~f:snd pair
;;
let unzip ?instrumentation ?left_result_equal ?right_result_equal input =
let data_equal =
Option.map2 left_result_equal right_result_equal ~f:(fun l r ->
Tuple2.equal ~eq1:l ~eq2:r)
in
unzip_mapi
?instrumentation
?data_equal
?left_result_equal
?right_result_equal
input
~f:(fun ~key:_ ~data -> data)
;;
let unzip_mapi_with_comparator'
(type v v1 v2 state_witness)
~instrumentation
?cutoff
?(data_equal = phys_equal)
(input : (('key, v, 'cmp) Map.t, state_witness) Incremental.t)
~(comparator : ('key, 'cmp) Comparator.t)
~(f :
key:'key
-> data:(v, state_witness) Incremental.t
-> (v1, state_witness) Incremental.t * (v2, state_witness) Incremental.t)
: (('key, v1, 'cmp) Map.t, state_witness) Incremental.t
* (('key, v2, 'cmp) Map.t, state_witness) Incremental.t
=
let module E = Incremental.Expert in
let incremental_state = Incremental.state input in
let empty_map = Map.Using_comparator.empty ~comparator in
let prev_map = ref empty_map in
let prev_nodes = ref empty_map in
let left_acc : ('key, v1, 'cmp) Map.t ref = ref empty_map in
let left_result = E.Node.create incremental_state (fun () -> !left_acc) in
let right_acc : ('key, v2, 'cmp) Map.t ref = ref empty_map in
let right_result = E.Node.create incremental_state (fun () -> !right_acc) in
let left_on_inner_change ~key data = left_acc := Map.set !left_acc ~key ~data in
let right_on_inner_change ~key data = right_acc := Map.set !right_acc ~key ~data in
let rec input_change =
lazy
(Incremental.map input ~f:(fun map ->
instrumentation.Instrumentation.f (fun () ->
let new_nodes =
Map.fold_symmetric_diff
~data_equal
!prev_map
map
~init:!prev_nodes
~f:(fun nodes (key, changed) ->
match changed with
| `Unequal _ ->
let node, _left_dep, _right_dep = Map.find_exn nodes key in
E.Node.make_stale node;
nodes
| `Left _ ->
let node, left_dep, right_dep = Map.find_exn nodes key in
let nodes = Map.remove nodes key in
E.Node.remove_dependency left_result left_dep;
E.Node.remove_dependency right_result right_dep;
left_acc := Map.remove !left_acc key;
right_acc := Map.remove !right_acc key;
E.Node.invalidate node;
nodes
| `Right _ ->
let node =
E.Node.create incremental_state (fun () ->
Map.find_exn !prev_map key)
in
Option.iter cutoff ~f:(fun c ->
Incremental.set_cutoff (E.Node.watch node) c);
E.Node.add_dependency node (E.Dependency.create (force input_change));
let left_incr, right_incr = f ~key ~data:(E.Node.watch node) in
let left_user_function_dep =
E.Dependency.create left_incr ~on_change:(left_on_inner_change ~key)
in
let right_user_function_dep =
E.Dependency.create
right_incr
~on_change:(right_on_inner_change ~key)
in
E.Node.add_dependency left_result left_user_function_dep;
E.Node.add_dependency right_result right_user_function_dep;
Map.set
nodes
~key
~data:(node, left_user_function_dep, right_user_function_dep))
in
prev_nodes := new_nodes;
prev_map := map)))
in
E.Node.add_dependency left_result (E.Dependency.create (force input_change));
E.Node.add_dependency right_result (E.Dependency.create (force input_change));
E.Node.watch left_result, E.Node.watch right_result
;;
let unzip_mapi' ?(instrumentation = no_instrumentation) ?cutoff ?data_equal map ~f =
let pair =
with_comparator map (fun comparator ->
map
|> unzip_mapi_with_comparator' ~instrumentation ?cutoff ?data_equal ~comparator ~f
|> Tuple2.uncurry Incremental.both)
in
Incremental.map ~f:fst pair, Incremental.map ~f:snd pair
;;
let keys ?(instrumentation = no_instrumentation) map =
with_comparator map (fun comparator ->
let add ~key ~data:_ acc = Set.add acc key in
let remove ~key ~data:_ acc = Set.remove acc key in
let data_equal _ _ = true in
unordered_fold
map
~instrumentation
~init:(Set.Using_comparator.empty ~comparator)
~revert_to_init_when_empty:true
~data_equal
~add
~remove)
;;
let partition_mapi ?(instrumentation = no_instrumentation) ?data_equal map ~f =
with_comparator map (fun comparator ->
let empty = Map.Using_comparator.empty ~comparator in
unordered_fold
?data_equal
map
~instrumentation
~init:(empty, empty)
~revert_to_init_when_empty:true
~update:(fun ~key ~old_data:_ ~new_data:data (first, second) ->
match f ~key ~data with
| First data -> Map.set first ~key ~data, Map.remove second key
| Second data -> Map.remove first key, Map.set second ~key ~data)
~add:(fun ~key ~data (first, second) ->
match f ~key ~data with
| First data -> Map.add_exn first ~key ~data, second
| Second data -> first, Map.add_exn second ~key ~data)
~remove:(fun ~key ~data:_ (first, second) ->
Map.remove first key, Map.remove second key))
;;
let partition_mapi' ?instrumentation ?cutoff ?data_equal map ~f =
mapi' ?instrumentation ?cutoff ?data_equal map ~f
|> partition_mapi ?instrumentation ~f:(fun ~key:_ ~data -> data)
;;
let flatten state map =
let module E = Incremental.Expert in
let result = ref (Map.Using_comparator.empty ~comparator:(Map.comparator map)) in
let node = E.Node.create state (fun () -> !result) in
Map.iteri map ~f:(fun ~key ~data:incr ->
E.Node.add_dependency
node
(E.Dependency.create incr ~on_change:(fun a ->
result := Map.set !result ~key ~data:a)));
E.Node.watch node
;;
let join_with_comparator ~instrumentation map_incr ~comparator =
let module E = Incremental.Expert in
let incremental_state = Incremental.state map_incr in
let empty_map = Map.Using_comparator.empty ~comparator in
let result_map = ref empty_map in
let old_map_of_incrs = ref empty_map in
let current_dependencies = ref empty_map in
let result = E.Node.create incremental_state (fun () -> !result_map) in
let add_subnode current_dependencies ~key ~data_node =
let new_dep =
E.Dependency.create data_node ~on_change:(fun data ->
result_map := Map.set !result_map ~key ~data)
in
E.Node.add_dependency result new_dep;
Map.set current_dependencies ~key ~data:new_dep
in
let remove_subnode current_dependencies ~key =
let dep = Map.find_exn current_dependencies key in
E.Node.remove_dependency result dep;
result_map := Map.remove !result_map key;
Map.remove current_dependencies key
in
let lhs_change =
Incremental.map map_incr ~f:(fun map_of_incrs ->
instrumentation.Instrumentation.f (fun () ->
let new_dependency_map =
Map.fold_symmetric_diff
~data_equal:phys_equal
!old_map_of_incrs
map_of_incrs
~init:!current_dependencies
~f:(fun current_dependencies (key, diff) ->
match diff with
| `Left _ -> remove_subnode current_dependencies ~key
| `Right data_node -> add_subnode current_dependencies ~key ~data_node
| `Unequal (_, data_node) ->
remove_subnode current_dependencies ~key |> add_subnode ~key ~data_node)
in
current_dependencies := new_dependency_map;
old_map_of_incrs := map_of_incrs))
in
E.Node.add_dependency result (E.Dependency.create lhs_change);
E.Node.watch result
;;
let join ?(instrumentation = no_instrumentation) map =
with_comparator map (fun comparator ->
join_with_comparator ~instrumentation map ~comparator)
;;
module Separate_state = struct
type ('k, 'v, 'cmp, 'w) t =
{ mutable input_map : ('k, 'v, 'cmp) Map.t
; mutable expert_nodes : ('k, ('v, 'w) Incremental.Expert.Node.t, 'cmp) Map.t
; mutable output_map : ('k, ('v, 'w) Incremental.t, 'cmp) Map.t
}
let create comparator =
let empty = Map.Using_comparator.empty ~comparator in
{ input_map = empty; expert_nodes = empty; output_map = empty }
;;
let create_lookup_node state t key =
Incremental.Expert.Node.create state (fun () -> Map.find_exn t.input_map key)
;;
end
let separate ?(instrumentation = no_instrumentation) input_map ~data_equal =
let incremental_state = Incremental.state input_map in
with_comparator input_map (fun comparator ->
let state = Separate_state.create comparator in
let output_map_node =
Incremental.Expert.Node.create incremental_state (fun () -> state.output_map)
in
let make_node_depend_on_input_map_changed node ~input_map_changed =
let dependency =
Incremental.Expert.Dependency.create (Lazy.force_val input_map_changed)
in
Incremental.Expert.Node.add_dependency node dependency
in
(* We want to make nodes depend on [input_map_changed] so that [input_map_changed]
is allowed to make them stale, but we do not want them to be recomputed for any
other reason. So we make [input_map_changed] a unit incremental (that therefore
never changes) and this way [output_map_node] and the lookup nodes will only be
recomputed when they are explicitly made stale.
*)
let rec input_map_changed =
lazy
(Incremental.map input_map ~f:(fun input_map ->
instrumentation.Instrumentation.f (fun () ->
let prev_input_map = state.input_map in