flan/test/test_agent.ml

(* The agent, end to end (NEXT.md, dev loop step 3).

   test_reload.ml proves the primitive with a C harness driving it. This one
   proves the thing the dev loop actually is: a program that is running its own
   loop, a redefinition arriving over a socket while it runs, and the swap
   becoming visible at a point the program chose.

   The split the agent exists for is between two threads. [dlopen] relocates a
   module and takes the loader lock — milliseconds, unbounded — so it happens
   on the listener thread. [flan_reload_install] is one store per function, and
   it must not land while a redefined function is on the stack, so it happens
   on the game thread when it asks. Everything here is arranged to make that
   observable rather than to make it fast. *)

open Flan

let failures = ref 0
let fail fmt = Printf.ksprintf (fun s -> incr failures; print_endline ("FAIL  " ^ s)) fmt

let scratch = Filename.get_temp_dir_name ()
let tmp name = Filename.concat scratch ("flan-agent-" ^ name)

(* Poll for a condition rather than sleeping a fixed time: the program has to
   bind its socket before there is anything to connect to, and how long that
   takes is not ours to predict. *)
let rec await ?(ms = 3000) f =
  if f () then true
  else if ms <= 0 then false
  else begin
    ignore (Unix.select [] [] [] 0.005);
    await ~ms:(ms - 5) f
  end

(* The socket file appears at [bind], which is a moment before [listen], so a
   connect can lose that race and get ECONNREFUSED. Retry rather than sleep. *)
let rec connect ?(ms = 2000) path =
  let s = Unix.socket Unix.PF_UNIX Unix.SOCK_STREAM 0 in
  match Unix.connect s (Unix.ADDR_UNIX path) with
  | () -> s
  | exception Unix.Unix_error (Unix.ECONNREFUSED, _, _) when ms > 0 ->
    Unix.close s;
    ignore (Unix.select [] [] [] 0.005);
    connect ~ms:(ms - 5) path

let send path line =
  let s = connect path in
  let msg = line ^ "\n" in
  ignore (Unix.write_substring s msg 0 (String.length msg));
  (* Read to EOF, not once: a reply arrives in several pieces, and closing
     after the first one is what made the agent take SIGPIPE. *)
  let buf = Bytes.create 512 in
  let b = Buffer.create 512 in
  let rec drain () =
    match Unix.read s buf 0 512 with
    | 0 -> ()
    | n -> Buffer.add_subbytes b buf 0 n; drain ()
    | exception Unix.Unix_error _ -> ()
  in
  drain ();
  Unix.close s;
  Buffer.contents b

let () =
  match Sys.command "command -v clang > /dev/null 2>&1 && command -v llc > /dev/null 2>&1" with
  | 0 ->
    (* The session is the program the process is about to be built from. Going
       through it rather than calling Emit directly is the point: it is what
       knows [tick] is a name the host has, so the module binds to its cell as
       a symbol instead of inventing a registry entry nobody publishes. *)
    let t, l = Session.create ~file:"programs/agent.flan" in

    (* A dev build, because that is what has cells to install into and exports
       them. The agent's own C and its -lpthread come from the package. *)
    let dev = { Build.default with Build.dev = true } in
    let exe = tmp "prog" in
    ignore
      (Build.executable ~opts:dev ~csrcs:l.Load.csrcs ~lflags:l.Load.lflags
         t.Session.host ~out:exe);

    (* One form, which is what C-c C-c sends. *)
    let c = Session.eval t "(defn tick [] i64 (set ticks (+ ticks 1000)) ticks)" in
    let so = tmp "tick.so" in
    ignore (Build.shared ~opts:dev ~ir:c.Session.ir ~out:so ());

    let sock = tmp "sock" in
    let out = tmp "out" in
    (try Sys.remove sock with Sys_error _ -> ());
    let fd = Unix.openfile out [ Unix.O_WRONLY; Unix.O_CREAT; Unix.O_TRUNC ] 0o600 in
    let pid =
      Unix.create_process exe [| exe; sock |] Unix.stdin fd Unix.stderr
    in
    Unix.close fd;

    if not (await (fun () -> Sys.file_exists sock)) then begin
      fail "the program never bound its socket";
      (try Unix.kill pid Sys.sigkill with Unix.Unix_error _ -> ())
    end else begin
      (* Refusing junk comes first, while the program is still running: the
         daemon is a separate process and can send anything, and a bad path
         must not take down the program it was sent to. Nothing is queued by
         it, so the program is still waiting afterwards. *)
      let reply = send sock "/nonexistent/nope.so" in
      if String.length reply < 4 || String.sub reply 0 4 <> "err " then
        fail "a bad path was not refused: %S" reply;

      (* "ok" means queued, not installed — the store happens on the other
         thread, at a time this one does not choose. *)
      let reply = send sock so in
      if reply <> "ok\n" then fail "agent replied %S, wanted \"ok\\n\"" reply;
      let _, status = Unix.waitpid [] pid in
      let text = In_channel.with_open_bin out In_channel.input_all in
      (* 1 from the original [tick], then 1001: the same call site, in a
         program that never stopped, running a body that did not exist when it
         started. *)
      if status <> Unix.WEXITED 0 || text <> "1\n1001\n" then
        fail "agent reload\n        got: %S (%s)\n     wanted: %S" text
          (match status with
           | Unix.WEXITED c -> Printf.sprintf "exit %d" c
           | Unix.WSIGNALED c -> Printf.sprintf "signal %d" c
           | Unix.WSTOPPED c -> Printf.sprintf "stopped %d" c)
          "1\n1001\n"
    end;

    List.iter (fun f -> try Sys.remove f with Sys_error _ -> ())
      [ exe; so; sock; out ];
    if !failures = 0 then print_endline "agent: all tests passed"
    else begin
      Printf.printf "\n%d failure(s)\n" !failures;
      exit 1
    end
  | _ -> print_endline "agent: skipped (no clang or llc on PATH)"