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 let map f xs =
  25   let xs = List.map (
  26     fun x ->
  27       let rfd, wfd = pipe () in
  28       (x, rfd, wfd)
  29   ) xs in
  30
  31   let xs = List.map (
  32     fun (x, rfd, wfd) ->
  33       match fork () with
  34       | 0 ->                            (* child *)
  35         close rfd;
  36         let y = Printexc.catch f x in
  37         (* Write the final value to the pipe. *)
  38         output_value (out_channel_of_descr wfd) y;
  39         exit 0
  40
  41       | pid ->                          (* parent *)
  42         close wfd;
  43         (pid, rfd)
  44   ) xs in
  45
  46   let errors = ref 0 in
  47   let xs = List.map (
  48     fun (pid, rfd) ->
  49       (* Read all the output from the pipe. *)
  50       let buf = Buffer.create 13 in
  51       let bytes = Bytes.create 4096 in
  52       let rec loop () =
  53         let len = read rfd bytes 0 (Bytes.length bytes) in
  54         if len > 0 then (
  55           Buffer.add_subbytes buf bytes 0 len;
  56           loop ()
  57         )
  58       in
  59       loop ();
  60       let str = Buffer.to_bytes buf in
  61
  62       (* Wait for the subprocess. *)
  63       match waitpid [] pid with
  64       | _, WEXITED 0 -> str
  65       | pid, WEXITED i ->
  66         eprintf "mclu: subprocess pid %d died with exit status %d\n" pid i;
  67         incr errors;
  68         Bytes.empty
  69       | pid, WSIGNALED i ->
  70         eprintf "mclu: subprocess pid %d died with signal %d\n" pid i;
  71         incr errors;
  72         Bytes.empty
  73       | pid, WSTOPPED i ->
  74         eprintf "mclu: subprocess pid %d stopped with signal %d\n" pid i;
  75         incr errors;
  76         Bytes.empty
  77   ) xs in
  78
  79   if !errors > 0 then
  80     exit 1;
  81
  82   xs
  83
  84 let iter f xs = ignore (map f xs)