-
Notifications
You must be signed in to change notification settings - Fork 20
/
parmap.ml
780 lines (693 loc) · 26.6 KB
/
parmap.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
(**************************************************************************)
(* ParMap: a simple library to perform Map computations on a multi-core *)
(* *)
(* Author(s): Marco Danelutto, Roberto Di Cosmo *)
(* *)
(* This library is free software: you can redistribute it and/or modify *)
(* it under the terms of the GNU Lesser General Public License as *)
(* published by the Free Software Foundation, either version 2 of the *)
(* License, or (at your option) any later version. A special linking *)
(* exception to the GNU Lesser General Public License applies to this *)
(* library, see the LICENSE file for more information. *)
(**************************************************************************)
module Utils = Parmap_utils
(* OS related constants *)
(* sequence type, subsuming lists and arrays *)
type 'a sequence = L of 'a list | A of 'a array
let debug_enabled = ref false
(* toggle debugging *)
let debugging b = debug_enabled:=b
(* default number of cores, and a setter function *)
let default_ncores=ref (max 2 (Setcore.numcores()-1));;
let set_default_ncores n = default_ncores := n;;
let get_default_ncores () = !default_ncores;;
let ncores = ref 0;;
let set_ncores n = ncores := n;;
let get_ncores () = !ncores
(* core mapping *)
let no_core_pinning = ref false
let disable_core_pinning () =
no_core_pinning := true
let enable_core_pinning () =
no_core_pinning := false
let core_mapping = ref None
let set_core_mapping (m: int array) = core_mapping := Some m
(* worker process rank *)
let masters_rank = -1
let rank = ref masters_rank
let set_rank n = rank := n
let get_rank () = !rank
(* exception handling code *)
let handle_exc core msg =
Utils.log_error "aborting due to exception on core %d: %s" core msg; exit 1;;
(* Helper functions for stdout/stderr redirection *)
let can_redirect path =
if not(Sys.file_exists path) then
try
Unix.mkdir path 0o777; true
with Unix.Unix_error(e,_s,_s') ->
(* another job may have created it between the check and the mkdir *)
if e == Unix.EEXIST then true
else begin
(Printf.eprintf "[Pid %d]: Error creating %s : %s; proceeding \
without stdout/stderr redirection\n%!"
(Unix.getpid ()) path (Unix.error_message e));
false
end
else true
let log_debug fmt =
Printf.ksprintf (
if !debug_enabled then begin
(fun s -> Format.eprintf "[Parmap]: %s@." s)
end else ignore
) fmt
(* freopen emulation, from Xavier's suggestion on OCaml mailing list *)
let reopen_out outchan path fname =
if can_redirect path then
begin
flush outchan;
let filename = Filename.concat path fname in
let fd1 = Unix.descr_of_out_channel outchan in
let fd2 = Unix.openfile
filename [Unix.O_WRONLY; Unix.O_CREAT; Unix.O_TRUNC] 0o666 in
Unix.dup2 fd2 fd1;
Unix.close fd2
end
else ()
(* send stdout and stderr to a file to avoid mixing output from different
cores, if enabled *)
let[@warning "-16" (*unerasable-optional-argument*)]
redirect ?(path = (Printf.sprintf "/tmp/.parmap.%d" (Unix.getpid ()))) ~id =
reopen_out stdout path (Printf.sprintf "stdout.%d" id);
reopen_out stderr path (Printf.sprintf "stderr.%d" id);;
(* unmarshal from a mmap seen as a bigarray *)
let unmarshal fd =
let a =
Bigarray.array1_of_genarray @@
Parmap_compat.map_file fd Bigarray.char Bigarray.c_layout true [|-1|] in
let res = Bytearray.unmarshal a 0 in
Unix.close fd;
res
(* marshal to a mmap seen as a bigarray *)
(* System dependent notes:
(* a reasonable size for mmapping a file containing even huge result data *)
let huge_size = if Sys.word_size = 64 then 1 lsl 32 else 1 lsl 26
- on Linux kernels, we might allocate a mmapped memory area of huge_size
and marshal into it directly
let ba =
Bigarray.array1_of_genarray @@
Parmap_compat.map_file fd Bigarray.char Bigarray.c_layout true huge_size in
ignore(Bytearray.marshal_to_buffer ba 0 v [Marshal.Closures]);
Unix.close fd
- to be compatible with other systems, notably Mac OS X, which insist in
allocating *all*
the declared memory area even for a sparse file, we must choose a less
efficient approach:
* marshal the value v to a string s, and compute its size
* allocate a mmap of that exact size,
* copy the string to that mmap
this allocates twice as much memory, and incurs an extra copy of the
value v
*)
let marshal fd v =
let s = Marshal.to_string v [Marshal.Closures] in
ignore(Bytearray.mmap_of_string fd s)
(* Exit the program with calling [at_exit] handlers *)
external sys_exit : int -> 'a = "caml_sys_exit"
let spawn_many n ~in_subprocess =
let rec loop i acc =
if i = n then
acc
else
match Unix.fork() with
0 ->
(* [at_exit] handlers are called in reverse order of registration.
By registering a handler that exits prematurely, we prevent the
execution of handlers registered before the fork.
This ignores the exit code provided by the user, but we ignore
it anyway in [wait_for_pids].
*)
at_exit (fun () -> sys_exit 0);
set_rank i;
in_subprocess i;
exit 0
| -1 ->
Utils.log_error "fork error: pid %d; i=%d" (Unix.getpid()) i;
loop (i + 1) acc
| pid ->
loop (i + 1) (pid :: acc)
in
(* call the GC before forking *)
Gc.compact ();
loop 0 []
let wait_for_pids pids =
let rec wait_for_pid pid =
try ignore(Unix.waitpid [] pid : int * Unix.process_status)
with
| Unix.Unix_error (Unix.ECHILD, _, _) -> ()
| Unix.Unix_error (Unix.EINTR, _, _) -> wait_for_pid pid
in
List.iter wait_for_pid pids
let run_many n ~in_subprocess =
wait_for_pids (spawn_many n ~in_subprocess)
(* a simple mapper function that computes 1/nth of the data on each of the n
cores in one iteration *)
let simplemapper (init:int -> unit) (finalize: unit -> unit) ncores' compute opid al collect =
(* flush everything *)
flush_all();
(* init task parameters *)
let ln = Array.length al in
set_ncores (min ln (max 1 ncores'));
let chunksize = max 1 (ln / !ncores) in
log_debug
"simplemapper on %d elements, on %d cores, chunksize = %d%!"
ln !ncores chunksize;
(* create descriptors to mmap *)
let fdarr=Array.init !ncores (fun _ -> Utils.tempfd()) in
let statusfdarr=Array.init !ncores (fun _ -> Utils.tempfd()) in
(* run children *)
run_many !ncores ~in_subprocess:(fun i ->
init i; (* call initialization function *)
at_exit finalize; (* register finalization function *)
let lo=i*chunksize in
let hi=if i = !ncores - 1 then ln - 1 else (i + 1) * chunksize - 1 in
let exc_handler e j = (* handle an exception at index j *)
Utils.log_error
"error at index j=%d in (%d,%d), chunksize=%d of a total of \
%d got exception %s on core %d \n%!"
j lo hi chunksize (hi-lo+1) (Printexc.to_string e) i;
marshal statusfdarr.(i) false;
exit 1
in
let v = compute al lo hi opid exc_handler in
marshal statusfdarr.(i) true;
marshal fdarr.(i) v);
(* read in all data *)
let res = ref [] in
let success = ref true in
(* check whether an exception has been raised in the subprocesses *)
for i = 0 to !ncores - 1 do
success:= !success && ((unmarshal statusfdarr.(i)):bool);
done;
if not !success then failwith "Aborting computation due to exception(s) raised in the workers";
(* iterate in reverse order, to accumulate in the right order *)
for i = 0 to !ncores - 1 do
res:= ((unmarshal fdarr.((!ncores-1)-i)):'d)::!res;
done;
(* collect all results *)
collect !res
(* a simple iteration function that iterates on 1/nth of the data on each of
the n cores *)
let simpleiter init finalize ncores' compute al =
(* flush everything *)
flush_all();
(* init task parameters *)
let ln = Array.length al in
set_ncores (min ln (max 1 ncores'));
let chunksize = max 1 (ln / !ncores) in
log_debug
"simplemapper on %d elements, on %d cores, chunksize = %d%!"
ln !ncores chunksize;
(* run children *)
run_many !ncores ~in_subprocess:(fun i ->
init i; (* call initialization function *)
at_exit finalize; (* register finalization function *)
let lo=i*chunksize in
let hi=if i= !ncores - 1 then ln-1 else (i+1)*chunksize-1 in
let exc_handler e j = (* handle an exception at index j *)
Utils.log_error
"error at index j=%d in (%d,%d), chunksize=%d of a total of \
%d got exception %s on core %d \n%!"
j lo hi chunksize (hi-lo+1) (Printexc.to_string e) i;
exit 1
in
compute al lo hi exc_handler);
(* return with no value *)
(* a more sophisticated mapper function, with automatic load balancing *)
(* the type of messages exchanged between master and workers *)
type msg_to_master = Ready of int | Error of int * string
type msg_to_worker = Finished | Task of int
let setup_children_chans oc pipedown ?fdarr i =
(if !no_core_pinning then ()
else match !core_mapping with
(* map process i to core i, or, if a core_mapping exist,
to core_mapping.(i), reusing core_mapping as many times as needed *)
| None -> Setcore.setcore i
| Some m ->
let ml = Array.length m in
Setcore.setcore m.(i mod ml));
(* close the other ends of the pipe and convert my ends to ic/oc *)
Unix.close (snd pipedown.(i));
let pid = Unix.getpid() in
let ic = Unix.in_channel_of_descr (fst pipedown.(i)) in
let receive () = Marshal.from_channel ic in
let signal v = Marshal.to_channel oc v []; flush oc in
let return v =
let d = Unix.gettimeofday() in
let _ = match fdarr with Some fdarr -> marshal fdarr.(i) v | None -> () in
log_debug "worker elapsed %f in marshalling" (Unix.gettimeofday() -. d) in
let finish () =
(log_debug "shutting down (pid=%d)\n%!" pid;
try close_in ic; close_out oc with _ -> ()
);
exit 0 in
receive, signal, return, finish, pid
(* parametric mapper primitive that captures the parallel structure *)
let mapper (init:int -> unit) (finalize:unit -> unit) ncores' ~chunksize compute opid al collect =
let ln = Array.length al in
if ln=0 then (collect []) else
begin
set_ncores (min ln (max 1 ncores'));
log_debug "mapper on %d elements, on %d cores%!" ln !ncores;
match chunksize with
None ->
(* no need of load balancing *)
simplemapper init finalize !ncores compute opid al collect
| Some v when !ncores >= ln/v ->
(* no need of load balancing if more cores than tasks *)
simplemapper init finalize !ncores compute opid al collect
| Some v ->
(* init task parameters : ntasks > 0 here,
as otherwise ncores >= 1 >= ln/v = ntasks and we would take
the branch above *)
let chunksize = v and ntasks = ln/v in
(* flush everything *)
flush_all ();
(* create descriptors to mmap *)
let fdarr=Array.init !ncores (fun _ -> Utils.tempfd()) in
(* setup communication channel with the workers *)
let pipedown=Array.init !ncores (fun _ -> Unix.pipe ()) in
let pipeup_rd,pipeup_wr=Unix.pipe () in
let oc_up = Unix.out_channel_of_descr pipeup_wr in
(* run children *)
let pids =
spawn_many !ncores ~in_subprocess:(fun i ->
init i; (* call initialization function *)
at_exit finalize; (* register finalization function *)
let d=Unix.gettimeofday() in
(* primitives for communication *)
Unix.close pipeup_rd;
let receive,signal,return,finish,pid =
setup_children_chans oc_up pipedown ~fdarr i in
let reschunk=ref opid in
let computetask n = (* compute chunk number n *)
let lo=n*chunksize in
let hi=if n=ntasks-1 then ln-1 else (n+1)*chunksize-1 in
let exc_handler e j = (* handle an exception at index j *)
begin
let errmsg = Printexc.to_string e in
Utils.log_error
"error at index j=%d in (%d,%d), chunksize=%d of a \
total of %d got exception %s on core %d \n%!"
j lo hi chunksize (hi-lo+1) errmsg i;
signal (Error (i,errmsg): msg_to_master);
finish()
end
in
reschunk:= compute al lo hi !reschunk exc_handler;
log_debug
"worker on core %d (pid=%d), segment (%d,%d) of data of \
length %d, chunksize=%d finished in %f seconds"
i pid lo hi ln chunksize (Unix.gettimeofday() -. d)
in
while true do
(* ask for work until we are finished *)
signal (Ready i);
match receive() with
| Finished -> return (!reschunk:'d); finish ()
| Task n -> computetask n
done)
in
(* close unused ends of the pipes *)
Array.iter (fun (rfd,_) -> Unix.close rfd) pipedown;
Unix.close pipeup_wr;
(* get ic/oc/wfdl *)
let ocs=
Array.init !ncores
(fun n -> Unix.out_channel_of_descr (snd pipedown.(n))) in
let ic=Unix.in_channel_of_descr pipeup_rd in
(* feed workers until all tasks are finished *)
for i=0 to ntasks-1 do
match Marshal.from_channel ic with
Ready w ->
(log_debug "sending task %d to worker %d" i w;
let oc = ocs.(w) in
(Marshal.to_channel oc (Task i) []); flush oc)
| (Error (core,msg): msg_to_master) -> handle_exc core msg
done;
(* send termination token to all children *)
Array.iter (fun oc ->
Marshal.to_channel oc Finished [];
flush oc;
close_out oc
) ocs;
(* wait for all children to terminate *)
wait_for_pids pids;
(* read in all data *)
let res = ref [] in
(* iterate in reverse order, to accumulate in the right order *)
for i = 0 to !ncores-1 do
res:= ((unmarshal fdarr.((!ncores-1)-i)):'d)::!res;
done;
(* close read pipe end *)
Unix.close pipeup_rd;
(* collect all results *)
collect !res
end
(* parametric iteration primitive that captures the parallel structure *)
let geniter init finalize ncores' ~chunksize compute al =
let ln = Array.length al in
if ln=0 then () else
begin
set_ncores (min ln (max 1 ncores'));
log_debug "geniter on %d elements, on %d cores%!" ln !ncores;
match chunksize with
None ->
simpleiter init finalize !ncores compute al (* no need of load balancing *)
| Some v when !ncores >= ln/v ->
simpleiter init finalize !ncores compute al (* no need of load balancing *)
| Some v ->
(* init task parameters *)
let chunksize = v and ntasks = ln/v in
(* flush everything *)
flush_all ();
(* setup communication channel with the workers *)
let pipedown=Array.init !ncores (fun _ -> Unix.pipe ()) in
let pipeup_rd,pipeup_wr=Unix.pipe () in
let oc_up = Unix.out_channel_of_descr pipeup_wr in
(* spawn children *)
let pids =
spawn_many !ncores ~in_subprocess:(fun i ->
init i; (* call initialization function *)
at_exit finalize; (* register finalization function *)
let d=Unix.gettimeofday() in
(* primitives for communication *)
Unix.close pipeup_rd;
let receive,signal,return,finish,pid =
setup_children_chans oc_up pipedown i in
let computetask n = (* compute chunk number n *)
let lo=n*chunksize in
let hi=if n=ntasks-1 then ln-1 else (n+1)*chunksize-1 in
let exc_handler e j = (* handle an exception at index j *)
begin
let errmsg = Printexc.to_string e in
Utils.log_error
"error at index j=%d in (%d,%d), chunksize=%d of \
a total of %d got exception %s on core %d \n%!"
j lo hi chunksize (hi-lo+1) errmsg i;
signal (Error (i,errmsg): msg_to_master);
finish()
end
in
compute al lo hi exc_handler;
log_debug
"worker on core %d (pid=%d), segment (%d,%d) of data \
of length %d, chunksize=%d finished in %f seconds"
i pid lo hi ln chunksize (Unix.gettimeofday() -. d)
in
while true do
(* ask for work until we are finished *)
signal (Ready i);
match receive() with
| Finished -> return(); finish ()
| Task n -> computetask n
done)
in
(* close unused ends of the pipes *)
Array.iter (fun (rfd,_) -> Unix.close rfd) pipedown;
Unix.close pipeup_wr;
(* get ic/oc/wfdl *)
let ocs=Array.init !ncores
(fun n -> Unix.out_channel_of_descr (snd pipedown.(n))) in
let ic=Unix.in_channel_of_descr pipeup_rd in
(* feed workers until all tasks are finished *)
for i=0 to ntasks-1 do
match Marshal.from_channel ic with
Ready w ->
(log_debug "sending task %d to worker %d" i w;
let oc = ocs.(w) in
(Marshal.to_channel oc (Task i) []); flush oc)
| (Error (core,msg): msg_to_master) -> handle_exc core msg
done;
(* send termination token to all children *)
Array.iter (fun oc ->
Marshal.to_channel oc Finished [];
flush oc;
close_out oc
) ocs;
(* wait for all children to terminate *)
wait_for_pids pids;
(* no data to return *)
end
(* the parallel mapfold function *)
let parmapifold
?(init = fun _ -> ())
?(finalize = fun () -> ())
?(ncores= !default_ncores)
?(chunksize)
(f:int -> 'a -> 'b)
(s:'a sequence)
(op:'b->'c->'c)
(opid:'c)
(concat:'c->'c->'c) : 'c=
(* enforce array to speed up access to the list elements *)
let al = match s with A al -> al | L l -> Array.of_list l in
let compute al lo hi previous exc_handler =
(* iterate in reverse order, to accumulate in the right order *)
let r = ref previous in
for j=0 to (hi-lo) do
try
let idx = hi-j in
r := op (f idx (Array.unsafe_get al idx)) !r;
with e -> exc_handler e j
done; !r
in
mapper
init finalize ncores ~chunksize compute opid al (fun r -> Utils.fold_right concat r opid)
let parmapfold
?(init = fun _ -> ())
?(finalize = fun () -> ())
?ncores
?(chunksize)
(f:'a -> 'b)
(s:'a sequence)
(op:'b->'c->'c)
(opid:'c)
(concat:'c->'c->'c) : 'c=
parmapifold ~init ~finalize ?ncores ?chunksize (fun _ x -> f x) s op opid concat
(* the parallel fold function *)
let parfold
?(init = fun _ -> ())
?(finalize = fun () -> ())
?(ncores= !default_ncores)
?chunksize
(op:'a -> 'b -> 'b)
(s:'a sequence)
(opid:'b)
(concat:'b->'b->'b) : 'b=
parmapfold ~init ~finalize ~ncores ?chunksize (fun x -> x) s op opid concat
(* the parallel map function *)
let mapi_range lo hi (f:int -> 'a -> 'b) a =
let l = hi-lo in
if l < 0 then [||] else begin
let r = Array.make (l+1) (f lo (Array.unsafe_get a lo)) in
for i = 1 to l do
let idx = lo+i in
Array.unsafe_set r i (f idx (Array.unsafe_get a idx))
done;
r
end
(* the parallel map function, on arrays *)
let array_parmapi
?(init = fun _ -> ())
?(finalize = fun () -> ())
?(ncores= !default_ncores)
?chunksize
?(keeporder=false)
(f:int -> 'a -> 'b)
(al:'a array) : 'b array=
(* compute, collect and opid definitions for reordering after load balancing *)
let compute_sorted a lo hi previous exc_handler =
try
(lo,mapi_range lo hi f a)::previous
with e -> exc_handler e lo
and collect_sorted (r:(int * 'b array) list list) =
let fragments = List.flatten r in
let ordered=List.map snd (List.stable_sort (fun (n,_) (m,_) -> n-m) fragments) in
Array.concat ordered
and opid_sorted = [(0,[||])]
(* compute, collect and opid definitions without reordering *)
and compute a lo hi previous exc_handler =
try
Array.concat [(mapi_range lo hi f a);previous]
with e -> exc_handler e lo
and collect r = Array.concat r
and opid = [||] in
let ln = Array.length al in
match keeporder, chunksize with
| _ , None ->
(* no need of load balancing *)
mapper init finalize ncores ~chunksize compute opid al collect
| _ , Some v when ncores >= ln/v ->
(* no need of load balancing if more cores than tasks *)
mapper init finalize ncores ~chunksize compute opid al collect
| false , Some _ ->
(* load balancing without reordering *)
mapper init finalize ncores ~chunksize compute opid al collect
| true , Some _ ->
(* load balancing with reordering *)
mapper init finalize ncores ~chunksize compute_sorted opid_sorted al collect_sorted
let array_parmap ?init ?finalize ?ncores ?chunksize ?keeporder (f:'a -> 'b) (al:'a array) : 'b array=
array_parmapi ?init ?finalize ?ncores ?chunksize ?keeporder (fun _ x -> f x) al
let parmapi
?(init = fun _ -> ())
?(finalize = fun () -> ())
?(ncores= !default_ncores)
?chunksize
?(keeporder=false)
(f:int ->'a -> 'b)
(s:'a sequence) : 'b list=
(* enforce array to speed up access to the list elements *)
let al = match s with A al -> al | L l -> Array.of_list l in
(* compute, collect and opid definitions without reordering *)
let compute al lo hi previous exc_handler =
(* iterate in reverse order, to accumulate in the right order,
and add to acc *)
let f' j =
try let idx = lo+j in f idx (Array.unsafe_get al idx)
with e -> exc_handler e j in
let rec aux acc =
function
0 -> (f' 0)::acc
| n -> aux ((f' n)::acc) (n-1)
in aux previous (hi-lo)
and collect r = Utils.concat_tr r
and opid = [] in
let ln = Array.length al in
match keeporder, chunksize with
_ , None ->
(* no need of load balancing *)
mapper init finalize ncores ~chunksize compute opid al collect
| _ , Some v when ncores >= ln/v ->
(* no need of load balancing if more cores than tasks *)
mapper init finalize ncores ~chunksize compute opid al collect
| false , Some _ ->
(* load balancing without reordering *)
mapper init finalize ncores ~chunksize compute opid al collect
| true , Some _ ->
(* load balancing with reordering *)
Array.to_list (array_parmapi ~init ~finalize ~ncores ?chunksize ~keeporder f al)
let parmap ?init ?finalize ?ncores ?chunksize ?keeporder (f:'a -> 'b) (s:'a sequence) : 'b list=
parmapi ?init ?finalize ?ncores ?chunksize ?keeporder (fun _ x -> f x) s
(* This code is highly optimised for operations on float arrays:
- knowing in advance the size of the result allows to
pre-allocate it in a shared memory space as a Bigarray;
- to write in the Bigarray memory area using the unsafe
functions for Arrays, we trick the OCaml compiler into
using the Bigarray memory as an Array as follows
Array.unsafe_get (Obj.magic arr_out) 1
This works because OCaml compiles access to float arrays
as unboxed data, without further integrity checks;
- the final copy into a real OCaml array is done via a memcpy in C.
This approach gives a performance which is 2 to 3 times higher
w.r.t. array_parmap, at the price of using Obj.magic and
knowledge on the internal representation of arrays and bigarrays.
*)
exception WrongArraySize
type buf=
(float, Bigarray.float64_elt, Bigarray.c_layout) Bigarray.Array1.t *
int;; (* should be a long int some day *)
let init_shared_buffer a =
let size = Array.length a in
let fd = Utils.tempfd() in
let arr =
Bigarray.array1_of_genarray @@
Parmap_compat.map_file fd Bigarray.float64 Bigarray.c_layout true [|size|] in
(* The mmap() function shall add an extra reference to the file associated
with the file descriptor fildes which is not removed by a subsequent
close() on that file descriptor.
http://pubs.opengroup.org/onlinepubs/009695399/functions/mmap.html
*)
Unix.close fd; (arr,size)
let array_float_parmapi
?(init = fun _ -> ())
?(finalize = fun () -> ())
?(ncores= !default_ncores)
?chunksize
?result
?sharedbuffer
(f:int -> 'a -> float)
(al:'a array) : float array =
let size = Array.length al in
if size=0 then [| |] else
begin
let barr_out =
match sharedbuffer with
Some (arr,s) ->
if s<size then
(Utils.log_error
"shared buffer is too small to hold the input in \
array_float_parmap";
raise WrongArraySize)
else arr
| None -> fst (init_shared_buffer al)
in
(* trick the compiler into accessing the Bigarray memory area as a float
array: the data in Bigarray is placed at offset 1 w.r.t. a normal array,
so we get a pointer to that zone into arr_out_as_array, and have it typed
as a float array *)
(*let barr_out_as_array = Array.unsafe_get (Obj.magic barr_out) 1 in*)
let compute _ lo hi _ exc_handler =
try
for i=lo to hi do
Bigarray.Array1.unsafe_set barr_out i (f i (Array.unsafe_get al i))
done
with e -> exc_handler e lo
in
mapper init finalize ncores ~chunksize compute () al (fun _r -> ());
let res =
match result with
None -> Bytearray.to_floatarray barr_out size
| Some a ->
if Array.length a < size then
(Utils.log_error
"result array is too small to hold the result in \
array_float_parmap";
raise WrongArraySize)
else
Bytearray.to_this_floatarray a barr_out size
in res
end
let array_float_parmap
?(init = fun _ -> ())
?(finalize = fun () -> ())
?ncores
?chunksize
?result
?sharedbuffer
(f:'a -> float)
(al:'a array) : float array =
array_float_parmapi
~init ~finalize ?ncores ?chunksize ?result ?sharedbuffer (fun _ x -> f x) al
(* the parallel iteration function *)
let pariteri
?(init = fun _ -> ())
?(finalize = fun () -> ())
?(ncores= !default_ncores)
?chunksize
(f:int -> 'a -> unit)
(s:'a sequence) : unit=
(* enforce array to speed up access to the list elements *)
let al = match s with A al -> al | L l -> Array.of_list l in
let compute al lo hi exc_handler =
(* iterate on the given segment *)
let f' j =
try let idx = lo+j in f idx (Array.unsafe_get al idx)
with e -> exc_handler e j in
for i = 0 to hi-lo do
f' i
done
in
geniter init finalize ncores ~chunksize compute al
let pariter ?init ?finalize ?ncores ?chunksize (f:'a -> unit) (s:'a sequence) : unit=
pariteri ?init ?finalize ?ncores ?chunksize (fun _ x -> f x) s