edk2.git no longer exists.
[mclu.git] / parallel.ml
1 (* mclu: Mini Cloud
2  * Copyright (C) 2014-2015 Red Hat Inc.
3  *
4  * This program is free software; you can redistribute it and/or modify
5  * it under the terms of the GNU General Public License as published by
6  * the Free Software Foundation; either version 2 of the License, or
7  * (at your option) any later version.
8  *
9  * This program is distributed in the hope that it will be useful,
10  * but WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12  * GNU General Public License for more details.
13  *
14  * You should have received a copy of the GNU General Public License
15  * along with this program; if not, write to the Free Software
16  * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17  *)
18
19 (* Simple forking parallel primitives. *)
20
21 open Printf
22 open Unix
23
24 (* Call _exit directly, ie. do not run OCaml atexit handlers. *)
25 external _exit : int -> 'a = "parallel_exit" "noalloc"
26
27 let map f xs =
28   let xs = List.map (
29     fun x ->
30       let rfd, wfd = pipe () in
31       (x, rfd, wfd)
32   ) xs in
33
34   let xs = List.map (
35     fun (x, rfd, wfd) ->
36       match fork () with
37       | 0 ->                            (* child *)
38         close rfd;
39         let y = Printexc.catch f x in
40         (* Write the final value to the pipe. *)
41         let chan = out_channel_of_descr wfd in
42         output_value chan y;
43         flush chan;
44         _exit 0
45
46       | pid ->                          (* parent *)
47         close wfd;
48         (pid, rfd)
49   ) xs in
50
51   let errors = ref 0 in
52   let xs = List.map (
53     fun (pid, rfd) ->
54       (* Read all the output from the pipe. *)
55       let buf = Buffer.create 13 in
56       let bytes = Bytes.create 4096 in
57       let rec loop () =
58         let len = read rfd bytes 0 (Bytes.length bytes) in
59         if len > 0 then (
60           Buffer.add_subbytes buf bytes 0 len;
61           loop ()
62         )
63       in
64       loop ();
65       let str = Buffer.to_bytes buf in
66
67       (* Wait for the subprocess. *)
68       match waitpid [] pid with
69       | _, WEXITED 0 -> str
70       | pid, WEXITED i ->
71         eprintf "mclu: subprocess pid %d died with exit status %d\n" pid i;
72         incr errors;
73         Bytes.empty
74       | pid, WSIGNALED i ->
75         eprintf "mclu: subprocess pid %d died with signal %d\n" pid i;
76         incr errors;
77         Bytes.empty
78       | pid, WSTOPPED i ->
79         eprintf "mclu: subprocess pid %d stopped with signal %d\n" pid i;
80         incr errors;
81         Bytes.empty
82   ) xs in
83
84   if !errors > 0 then
85     exit 1;
86
87   xs
88
89 let iter f xs = ignore (map f xs)