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)