<commit_before>(*
 * Copyright (c) Facebook, Inc. and its affiliates.
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 *)

(*****************************************************************************
 * The job executed by the worker.
 *
 * The 'serializer' is the job continuation: it is a function that must
 * be called at the end of the request ir order to send back the result
 * to the worker process (this is "internal business", this is not visible
 * outside this module). The clone process will provide the expected
 * function. cf 'send_result' in 'worker_main'.
 *
 *****************************************************************************)

type request = Request of (serializer -> unit)

and serializer = { send: 'a. 'a -> unit }

type job_status = Job_terminated of Unix.process_status

let on_job_cancelled parent_outfd =
  (* The cancelling controller will ignore result of cancelled job anyway (see
   * wait_for_cancel function), so we can send back anything. Write twice, since
   * the normal response writes twice too *)
  Marshal_tools.to_fd_with_preamble parent_outfd "anything" |> ignore;
  Marshal_tools.to_fd_with_preamble parent_outfd "anything" |> ignore

(*****************************************************************************
 * Entry point for spawned worker.
 *
 *****************************************************************************)

let worker_main ic oc =
  let start_user_time = ref 0. in
  let start_system_time = ref 0. in
  let start_minor_words = ref 0. in
  let start_promoted_words = ref 0. in
  let start_major_words = ref 0. in
  let start_minor_collections = ref 0 in
  let start_major_collections = ref 0 in
  let start_wall_time = ref 0. in
  let start_proc_fs_status = ref None in
  let infd = Daemon.descr_of_in_channel ic in
  let outfd = Daemon.descr_of_out_channel oc in
  let send_result data =
    Mem_profile.stop ();
    let tm = Unix.times () in
    let end_user_time = tm.Unix.tms_utime +. tm.Unix.tms_cutime in
    let end_system_time = tm.Unix.tms_stime +. tm.Unix.tms_cstime in
    let {
      Gc.minor_words = end_minor_words;
      promoted_words = end_promoted_words;
      major_words = end_major_words;
      minor_collections = end_minor_collections;
      major_collections = end_major_collections;
      _;
    } =
      Gc.quick_stat ()
    in
    let (major_time, minor_time) = Sys_utils.get_gc_time () in
    Measure.sample "worker_gc_major_wall_time" major_time;
    Measure.sample "worker_gc_minor_wall_time" minor_time;

    Measure.sample "worker_user_time" (end_user_time -. !start_user_time);
    Measure.sample "worker_system_time" (end_system_time -. !start_system_time);
    Measure.sample "worker_wall_time" (Unix.gettimeofday () -. !start_wall_time);

    Measure.track_distribution "minor_words" ~bucket_size:(float (100 * 1024 * 1024));
    Measure.sample "minor_words" (end_minor_words -. !start_minor_words);

    Measure.track_distribution "promoted_words" ~bucket_size:(float (25 * 1024 * 1024));
    Measure.sample "promoted_words" (end_promoted_words -. !start_promoted_words);

    Measure.track_distribution "major_words" ~bucket_size:(float (50 * 1024 * 1024));
    Measure.sample "major_words" (end_major_words -. !start_major_words);

    Measure.sample "minor_collections" (float (end_minor_collections - !start_minor_collections));
    Measure.sample "major_collections" (float (end_major_collections - !start_major_collections));

    begin
      match (!start_proc_fs_status, ProcFS.status_for_pid (Unix.getpid ())) with
      | (Some { ProcFS.rss_total = start; _ }, Ok { ProcFS.rss_total = total; rss_hwm = hwm; _ }) ->
        Measure.sample "worker_rss_start" (float start);
        Measure.sample "worker_rss_delta" (float (total - start));
        Measure.sample "worker_rss_hwm_delta" (float (hwm - start))
      | _ -> ()
    end;

    (* If we got so far, just let it finish "naturally" *)
    WorkerCancel.set_on_worker_cancelled (fun () -> ());
    let len =
      Measure.time "worker_send_response" (fun () ->
          Marshal_tools.to_fd_with_preamble ~flags:[Marshal.Closures] outfd data)
    in
    if len > 30 * 1024 * 1024 (* 30 MB *) then (
      Hh_logger.log
        "WARNING: you are sending quite a lot of data (%d bytes), which may have an adverse performance impact. If you are sending closures, double-check to ensure that they have not captured large
        values in their environment."
        len;
      Printf.eprintf "%s" (Printexc.raw_backtrace_to_string (Printexc.get_callstack 100))
    );

    Measure.sample "worker_response_len" (float len);

    let stats = Measure.serialize (Measure.pop_global ()) in
    let _ = Marshal_tools.to_fd_with_preamble outfd stats in
    ()
  in
  try
    try
      Measure.push_global ();
      let (Request do_process) =
        Measure.time "worker_read_request" (fun () -> Marshal_tools.from_fd_with_preamble infd)
      in
      WorkerCancel.set_on_worker_cancelled (fun () -> on_job_cancelled outfd);
      let tm = Unix.times () in
      let gc = Gc.quick_stat () in
      Sys_utils.start_gc_profiling ();

      start_user_time := tm.Unix.tms_utime +. tm.Unix.tms_cutime;
      start_system_time := tm.Unix.tms_stime +. tm.Unix.tms_cstime;
      start_minor_words := gc.Gc.minor_words;
      start_promoted_words := gc.Gc.promoted_words;
      start_major_words := gc.Gc.major_words;
      start_minor_collections := gc.Gc.minor_collections;
      start_major_collections := gc.Gc.major_collections;
      start_wall_time := Unix.gettimeofday ();
      start_proc_fs_status := ProcFS.status_for_pid (Unix.getpid ()) |> Base.Result.ok;
      Mem_profile.start ();
      do_process { send = send_result };
      exit 0
    with
    | End_of_file -> exit 1
    | SharedMem.Out_of_shared_memory -> FlowExitStatus.(exit Out_of_shared_memory)
    | SharedMem.Hash_table_full -> FlowExitStatus.(exit Hash_table_full)
    | SharedMem.Heap_full -> FlowExitStatus.(exit Heap_full)
  with e ->
    let exn = Exception.wrap e in
    let e_str = Exception.get_ctor_string exn in
    let pid = Unix.getpid () in
    Printf.printf "Worker %d exception: %s\n%!" pid e_str;
    if EventLogger.should_log () then EventLogger.worker_exception e_str;
    Printf.printf
      "Worker %d Potential backtrace:\n%s\n%!"
      pid
      (Exception.get_full_backtrace_string max_int exn);
    exit 2

let win32_worker_main restore (state, _controller_fd) (ic, oc) =
  restore state;
  worker_main ic oc

let maybe_send_status_to_controller fd status =
  match fd with<commit_msg>exit workers gracefully when server dies<commit_after>(*
 * Copyright (c) Facebook, Inc. and its affiliates.
 *
 * This source code is licensed under the MIT license found in the
 * LICENSE file in the root directory of this source tree.
 *)

(*****************************************************************************
 * The job executed by the worker.
 *
 * The 'serializer' is the job continuation: it is a function that must
 * be called at the end of the request ir order to send back the result
 * to the worker process (this is "internal business", this is not visible
 * outside this module). The clone process will provide the expected
 * function. cf 'send_result' in 'worker_main'.
 *
 *****************************************************************************)

type request = Request of (serializer -> unit)

and serializer = { send: 'a. 'a -> unit }

type job_status = Job_terminated of Unix.process_status

exception Connection_closed

let on_job_cancelled parent_outfd =
  (* The cancelling controller will ignore result of cancelled job anyway (see
   * wait_for_cancel function), so we can send back anything. Write twice, since
   * the normal response writes twice too *)
  Marshal_tools.to_fd_with_preamble parent_outfd "anything" |> ignore;
  Marshal_tools.to_fd_with_preamble parent_outfd "anything" |> ignore

(*****************************************************************************
 * Entry point for spawned worker.
 *
 *****************************************************************************)

let worker_main ic oc =
  let start_user_time = ref 0. in
  let start_system_time = ref 0. in
  let start_minor_words = ref 0. in
  let start_promoted_words = ref 0. in
  let start_major_words = ref 0. in
  let start_minor_collections = ref 0 in
  let start_major_collections = ref 0 in
  let start_wall_time = ref 0. in
  let start_proc_fs_status = ref None in
  let infd = Daemon.descr_of_in_channel ic in
  let outfd = Daemon.descr_of_out_channel oc in
  let send_result data =
    Mem_profile.stop ();
    let tm = Unix.times () in
    let end_user_time = tm.Unix.tms_utime +. tm.Unix.tms_cutime in
    let end_system_time = tm.Unix.tms_stime +. tm.Unix.tms_cstime in
    let {
      Gc.minor_words = end_minor_words;
      promoted_words = end_promoted_words;
      major_words = end_major_words;
      minor_collections = end_minor_collections;
      major_collections = end_major_collections;
      _;
    } =
      Gc.quick_stat ()
    in
    let (major_time, minor_time) = Sys_utils.get_gc_time () in
    Measure.sample "worker_gc_major_wall_time" major_time;
    Measure.sample "worker_gc_minor_wall_time" minor_time;

    Measure.sample "worker_user_time" (end_user_time -. !start_user_time);
    Measure.sample "worker_system_time" (end_system_time -. !start_system_time);
    Measure.sample "worker_wall_time" (Unix.gettimeofday () -. !start_wall_time);

    Measure.track_distribution "minor_words" ~bucket_size:(float (100 * 1024 * 1024));
    Measure.sample "minor_words" (end_minor_words -. !start_minor_words);

    Measure.track_distribution "promoted_words" ~bucket_size:(float (25 * 1024 * 1024));
    Measure.sample "promoted_words" (end_promoted_words -. !start_promoted_words);

    Measure.track_distribution "major_words" ~bucket_size:(float (50 * 1024 * 1024));
    Measure.sample "major_words" (end_major_words -. !start_major_words);

    Measure.sample "minor_collections" (float (end_minor_collections - !start_minor_collections));
    Measure.sample "major_collections" (float (end_major_collections - !start_major_collections));

    begin
      match (!start_proc_fs_status, ProcFS.status_for_pid (Unix.getpid ())) with
      | (Some { ProcFS.rss_total = start; _ }, Ok { ProcFS.rss_total = total; rss_hwm = hwm; _ }) ->
        Measure.sample "worker_rss_start" (float start);
        Measure.sample "worker_rss_delta" (float (total - start));
        Measure.sample "worker_rss_hwm_delta" (float (hwm - start))
      | _ -> ()
    end;

    (* If we got so far, just let it finish "naturally" *)
    WorkerCancel.set_on_worker_cancelled (fun () -> ());
    let len =
      Measure.time "worker_send_response" (fun () ->
          try Marshal_tools.to_fd_with_preamble ~flags:[Marshal.Closures] outfd data
          with Unix.Unix_error (Unix.EPIPE, _, _) -> raise Connection_closed)
    in
    if len > 30 * 1024 * 1024 (* 30 MB *) then (
      Hh_logger.log
        "WARNING: you are sending quite a lot of data (%d bytes), which may have an adverse performance impact. If you are sending closures, double-check to ensure that they have not captured large
        values in their environment."
        len;
      Printf.eprintf "%s" (Printexc.raw_backtrace_to_string (Printexc.get_callstack 100))
    );

    Measure.sample "worker_response_len" (float len);

    let stats = Measure.serialize (Measure.pop_global ()) in
    let _ =
      try Marshal_tools.to_fd_with_preamble outfd stats
      with Unix.Unix_error (Unix.EPIPE, _, _) -> raise Connection_closed
    in
    ()
  in
  try
    try
      Measure.push_global ();
      let (Request do_process) =
        Measure.time "worker_read_request" (fun () ->
            try Marshal_tools.from_fd_with_preamble infd
            with End_of_file -> raise Connection_closed)
      in
      WorkerCancel.set_on_worker_cancelled (fun () -> on_job_cancelled outfd);
      let tm = Unix.times () in
      let gc = Gc.quick_stat () in
      Sys_utils.start_gc_profiling ();

      start_user_time := tm.Unix.tms_utime +. tm.Unix.tms_cutime;
      start_system_time := tm.Unix.tms_stime +. tm.Unix.tms_cstime;
      start_minor_words := gc.Gc.minor_words;
      start_promoted_words := gc.Gc.promoted_words;
      start_major_words := gc.Gc.major_words;
      start_minor_collections := gc.Gc.minor_collections;
      start_major_collections := gc.Gc.major_collections;
      start_wall_time := Unix.gettimeofday ();
      start_proc_fs_status := ProcFS.status_for_pid (Unix.getpid ()) |> Base.Result.ok;
      Mem_profile.start ();
      do_process { send = send_result };
      exit 0
    with
    | Connection_closed -> exit 1
    | SharedMem.Out_of_shared_memory -> FlowExitStatus.(exit Out_of_shared_memory)
    | SharedMem.Hash_table_full -> FlowExitStatus.(exit Hash_table_full)
    | SharedMem.Heap_full -> FlowExitStatus.(exit Heap_full)
  with e ->
    let exn = Exception.wrap e in
    let e_str = Exception.get_ctor_string exn in
    let pid = Unix.getpid () in
    Printf.printf "Worker %d exception: %s\n%!" pid e_str;
    if EventLogger.should_log () then EventLogger.worker_exception e_str;
    Printf.printf
      "Worker %d Potential backtrace:\n%s\n%!"
      pid
      (Exception.get_full_backtrace_string max_int exn);
    exit 2

let win32_worker_main restore (state, _controller_fd) (ic, oc) =
  restore state;
  worker_main ic oc

let maybe_send_status_to_controller fd status =
  match fd with
{'fraction_non_alphanumeric': 0.134394341290893, 'fraction_numerical': 0.008694370763336281, 'mean_word_length': 3.690048375950242, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3494766', 'n_tokens_mistral': 4563, 'n_tokens_neox': 4264, 'n_words': 1113}