/usr/lib/ocaml/oUnit/oUnitRunnerProcesses.ml is in libounit-ocaml-dev 2.0.0-2build1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | (**************************************************************************)
(* The OUnit library *)
(* *)
(* Copyright (C) 2002-2008 Maas-Maarten Zeeman. *)
(* Copyright (C) 2010 OCamlCore SARL *)
(* Copyright (C) 2013 Sylvain Le Gall *)
(* *)
(* The package OUnit is copyright by Maas-Maarten Zeeman, OCamlCore SARL *)
(* and Sylvain Le Gall. *)
(* *)
(* Permission is hereby granted, free of charge, to any person obtaining *)
(* a copy of this document and the OUnit software ("the Software"), to *)
(* deal in the Software without restriction, including without limitation *)
(* the rights to use, copy, modify, merge, publish, distribute, *)
(* sublicense, and/or sell copies of the Software, and to permit persons *)
(* to whom the Software is furnished to do so, subject to the following *)
(* conditions: *)
(* *)
(* The above copyright notice and this permission notice shall be *)
(* included in all copies or substantial portions of the Software. *)
(* *)
(* The Software is provided ``as is'', without warranty of any kind, *)
(* express or implied, including but not limited to the warranties of *)
(* merchantability, fitness for a particular purpose and noninfringement. *)
(* In no event shall Maas-Maarten Zeeman be liable for any claim, damages *)
(* or other liability, whether in an action of contract, tort or *)
(* otherwise, arising from, out of or in connection with the Software or *)
(* the use or other dealings in the software. *)
(* *)
(* See LICENSE.txt for details. *)
(**************************************************************************)
(** Use processes to run several tests in parallel.
*
* Run processes that handle running tests. The processes read test, execute
* it, and communicate back to the master the log.
*
* This need to be done in another process because ocaml Threads are not truly
* concurrent. Moreover we cannot use Unix.fork because it's not portable
*)
open OUnitLogger
open OUnitTest
open OUnitState
open Unix
open OUnitRunner.GenericWorker
(* Create functions to handle sending and receiving data over a file descriptor.
*)
let make_channel
shard_id
string_of_read_message
string_of_written_message
fd_read
fd_write =
let () =
set_nonblock fd_read;
set_close_on_exec fd_read;
set_close_on_exec fd_write
in
let chn_write = out_channel_of_descr fd_write in
let really_read fd str =
let off = ref 0 in
let read = ref 0 in
while !read < String.length str do
try
let one_read =
Unix.read fd str !off (String.length str - !off)
in
read := !read + one_read;
off := !off + one_read
with Unix_error(EAGAIN, _, _) ->
()
done;
str
in
let header_str = String.create Marshal.header_size in
let send_data msg =
Marshal.to_channel chn_write msg [];
Pervasives.flush chn_write
in
let receive_data () =
try
let data_size = Marshal.data_size (really_read fd_read header_str) 0 in
let data_str = really_read fd_read (String.create data_size) in
let msg = Marshal.from_string (header_str ^ data_str) 0 in
msg
with Failure(msg) ->
OUnitUtils.failwithf "Communication error with worker processes: %s" msg
in
let close () =
close_out chn_write;
in
wrap_channel
shard_id
string_of_read_message
string_of_written_message
{
send_data = send_data;
receive_data = receive_data;
close = close
}
let processes_grace_period =
OUnitConf.make_float
"processes_grace_period"
5.0
"Delay to wait for a process to stop."
let processes_kill_period =
OUnitConf.make_float
"processes_kill_period"
5.0
"Delay to wait for a process to stop after killing it."
let create_worker conf map_test_cases shard_id master_id worker_log_file =
let safe_close fd = try close fd with Unix_error _ -> () in
let pipe_read_from_worker, pipe_write_to_master = Unix.pipe () in
let pipe_read_from_master, pipe_write_to_worker = Unix.pipe () in
match Unix.fork () with
| 0 ->
(* Child process. *)
let () =
safe_close pipe_read_from_worker;
safe_close pipe_write_to_worker;
(* stdin/stdout/stderr remain open and shared with master. *)
()
in
let channel =
make_channel
shard_id
string_of_message_to_worker
string_of_message_from_worker
pipe_read_from_master
pipe_write_to_master
in
main_worker_loop
conf ignore channel shard_id map_test_cases worker_log_file;
channel.close ();
safe_close pipe_read_from_master;
safe_close pipe_write_to_master;
exit 0
| pid ->
let channel =
make_channel
master_id
string_of_message_from_worker
string_of_message_to_worker
pipe_read_from_worker
pipe_write_to_worker
in
let rstatus = ref None in
let msg_of_process_status status =
if status = WEXITED 0 then
None
else
Some (OUnitUtils.string_of_process_status status)
in
let is_running () =
match !rstatus with
| None ->
let pid, status = waitpid [WNOHANG] pid in
if pid <> 0 then begin
rstatus := Some status;
false
end else begin
true
end
| Some _ ->
false
in
let close_worker () =
let rec wait_end timeout =
if timeout < 0.0 then begin
false, None
end else begin
let running = is_running () in
if running then
(* Wait 0.1 seconds and continue. *)
let _, _, _ = Unix.select [] [] [] 0.1 in
wait_end (timeout -. 0.1)
else
match !rstatus with
| Some status -> true, msg_of_process_status status
| None -> true, None
end
in
let ended, msg_opt =
channel.close ();
safe_close pipe_read_from_worker;
safe_close pipe_write_to_worker;
(* Recovery for worker going wild and not dying. *)
List.fold_left
(fun (ended, msg_opt) signal ->
if ended then begin
ended, msg_opt
end else begin
kill pid signal;
wait_end (processes_kill_period conf)
end)
(wait_end (processes_grace_period conf))
[15 (* SIGTERM *); 9 (* SIGKILL *)]
in
if ended then
msg_opt
else
Some (Printf.sprintf "unable to kill process %d" pid)
in
{
channel = channel;
close_worker = close_worker;
select_fd = pipe_read_from_worker;
shard_id = shard_id;
is_running = is_running;
}
(* Filter running workers waiting data. *)
let workers_waiting workers timeout =
let workers_fd_lst =
List.rev_map (fun worker -> worker.select_fd) workers
in
let workers_fd_waiting_lst, _, _ =
Unix.select workers_fd_lst [] [] timeout
in
List.filter
(fun workers -> List.memq workers.select_fd workers_fd_waiting_lst)
workers
let init () =
if Sys.os_type = "Unix" then
OUnitRunner.register "processes" 100
(runner create_worker workers_waiting)
|