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 ->
let () =
safe_close pipe_read_from_worker;
safe_close pipe_write_to_worker;
()
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
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;
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 ; 9 ]
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;
}