Add support for pty forking

Based on Ryan's code in rc/aeon

lib/eio/lib_eio/unix/dune

@@ -4,5 +4,5 @@
  (foreign_stubs
   (language c)
   (include_dirs include)
-  (names fork_action stubs))
+  (names fork_action stubs pty))
  (libraries eio unix threads mtime.clock.os))

lib/eio/lib_eio/unix/eio_unix.ml

@@ -1,6 +1,7 @@
 [@@@alert "-unstable"]
 
 module Fd = Fd
+module Pty = Pty
 module Resource = Resource
 module Private = Private
 

lib/eio/lib_eio/unix/eio_unix.mli

@@ -105,6 +105,9 @@ val pipe : Switch.t -> source * sink
 module Process = Process
 (** Spawning child processes with extra control. *)
 
+module Pty = Pty
+(** Pseudoterminal handling functions. *)
+
 (** The set of resources provided to a process on a Unix-compatible system. *)
 module Stdenv : sig
   type base = <

lib/eio/lib_eio/unix/fork_action.c

@@ -3,6 +3,9 @@
 #include <fcntl.h>
 #include <string.h>
 #include <errno.h>
+#include <sys/ioctl.h>
+#include <paths.h>
+#include <utmp.h>
 
 #include <caml/mlvalues.h>
 
@@ -138,26 +141,26 @@ static void action_dups(int errors, value v_config) {
     if (dst == -1) {
       // Dup to a temporary FD
       if (tmp == -1) {
-	tmp = dup(src);
-	if (tmp < 0) {
-	  eio_unix_fork_error(errors, "dup-tmp", strerror(errno));
-	  _exit(1);
-	}
+        tmp = dup(src);
+        if (tmp < 0) {
+          eio_unix_fork_error(errors, "dup-tmp", strerror(errno));
+          _exit(1);
+        }
       } else {
-	int r = dup2(src, tmp);
-	if (r < 0) {
-	  eio_unix_fork_error(errors, "dup2-tmp", strerror(errno));
-	  _exit(1);
-	}
+        int r = dup2(src, tmp);
+        if (r < 0) {
+          eio_unix_fork_error(errors, "dup2-tmp", strerror(errno));
+          _exit(1);
+        }
       }
       set_cloexec(errors, tmp, 1);
     } else if (src == dst) {
       set_cloexec(errors, dst, 0);
     } else {
       int r = dup2(src, dst);
       if (r < 0) {
-	eio_unix_fork_error(errors, "dup2", strerror(errno));
-	_exit(1);
+        eio_unix_fork_error(errors, "dup2", strerror(errno));
+        _exit(1);
       }
     }
     v_plan = Field(v_plan, 1);
@@ -174,3 +177,14 @@ static void action_dups(int errors, value v_config) {
 CAMLprim value eio_unix_fork_dups(value v_unit) {
   return Val_fork_fn(action_dups);
 }
+
+static void action_login_tty(int errors, value v_config) {
+  value v_pty = Field(v_config, 1);
+
+  if (login_tty(Int_val(v_pty)) == -1)
+    dprintf(errors, "action_login_tty Error logging in tty: %s", strerror(errno));
+}
+
+CAMLprim value eio_unix_login_tty(value v_unit) {
+  return Val_fork_fn(action_login_tty);
+}

lib/eio/lib_eio/unix/fork_action.ml

@@ -63,3 +63,10 @@ let inherit_fds m =
   with_fds m @@ fun m ->
   let plan : action list = Inherit_fds.plan m in
   { run = fun k -> k (Obj.repr (action_dups, plan, blocking)) }
+
+external action_login_tty : unit -> fork_fn = "eio_unix_login_tty"
+let action_login_tty = action_login_tty ()
+
+let login_tty pty =
+  Fd.use_exn "login_tty" pty @@ fun pty ->
+  { run = fun k -> k (Obj.repr (action_login_tty, pty)) }

lib/eio/lib_eio/unix/fork_action.mli

@@ -58,3 +58,6 @@ val inherit_fds : (int * Fd.t * [< blocking]) list -> t
     A mapping from an FD to itself simply clears the close-on-exec flag.
 
     After this, the new FDs may also be set as blocking or non-blocking, depending on [flags]. *)
+
+val login_tty : Fd.t -> t
+(** [login_tty pty] prepares for a shell login on the [pty] file descriptor. *)

lib/eio/lib_eio/unix/process.ml

@@ -96,6 +96,7 @@ class virtual mgr = object (self)
       2, stderr_fd, `Blocking;
     ] in
     self#spawn_unix ~sw ?cwd ~env ~fds ~executable args
+
 end
 
 let spawn_unix ~sw (mgr:#mgr) ?cwd ~fds ?env ?executable args =

lib/eio/lib_eio/unix/pty.c

@@ -0,0 +1,87 @@
+/*
+ * Copyright (c) 2004 Anil Madhavapeddy <anil@recoil.org>
+ * Copyright (c) 2020–2021 Craig Ferguson <me@craigfe.io>
+ *
+ * Permission to use, copy, modify, and distribute this software for any
+ * purpose with or without fee is hereby granted, provided that the above
+ * copyright notice and this permission notice appear in all copies.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
+ * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
+ * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
+ * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
+ * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
+ * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
+ * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+ */
+
+
+#include <stdio.h>
+#include <errno.h>
+#include <paths.h>
+#include <fcntl.h>
+#include <string.h>
+#include <termios.h>
+#include <unistd.h>
+#include <sys/ioctl.h>
+#include <signal.h>
+#include <limits.h>
+
+#include <pty.h>
+#include <utmp.h>
+
+#include <caml/mlvalues.h>
+#include <caml/alloc.h>
+#include <caml/memory.h>
+#include <caml/fail.h>
+#include <caml/callback.h>
+#include <caml/signals.h>
+#include <caml/unixsupport.h>
+
+value eio_unix_open_pty(value v_unit)
+{
+    CAMLparam1 (v_unit);
+    char namebuf[4096]; /* Not PATH_MAX due to portability issues */
+    int i, masterfd, slavefd;
+    CAMLlocal1(v_ret);
+
+    i = openpty(&masterfd, &slavefd, namebuf, NULL, NULL);
+    if (i < 0)
+        caml_uerror("openpty", Nothing);
+
+    v_ret = caml_alloc_small(3, 0);
+    Store_field(v_ret, 0, Val_int(masterfd));
+    Store_field(v_ret, 1, Val_int(slavefd));
+    Store_field(v_ret, 2, caml_copy_string(namebuf));
+    CAMLreturn (v_ret);
+}
+
+value eio_unix_window_size(value pty, value pty_window)
+{
+    CAMLparam2 (pty, pty_window);
+    int ptyfd;
+    struct winsize w;
+    w.ws_row = Int32_val(Field(pty_window, 0));
+    w.ws_col = Int32_val(Field(pty_window, 1));
+    w.ws_xpixel = Int32_val(Field(pty_window, 2));
+    w.ws_ypixel = Int32_val(Field(pty_window, 3));
+    ptyfd = Int_val(Field(pty, 0));
+    ioctl(ptyfd, TIOCSWINSZ, &w);
+    CAMLreturn (Val_unit);
+}
+
+value eio_unix_tty_window_size(value unit)
+{
+    CAMLparam1 (unit);
+    CAMLlocal1(pty_window);
+
+    struct winsize w;
+    if (ioctl(STDOUT_FILENO, TIOCGWINSZ, &w) == -1)
+        memset(&w, 0, sizeof(w));
+    pty_window = caml_alloc_small(4, 0);
+    Store_field(pty_window, 0, caml_copy_int32(w.ws_row));
+    Store_field(pty_window, 1, caml_copy_int32(w.ws_col));
+    Store_field(pty_window, 2, caml_copy_int32(w.ws_xpixel));
+    Store_field(pty_window, 3, caml_copy_int32(w.ws_ypixel));
+    CAMLreturn (pty_window);
+}

lib/eio/lib_eio/unix/pty.ml

@@ -0,0 +1,16 @@
+type pty = {
+  masterfd : Unix.file_descr;
+  slavefd : Unix.file_descr;
+  name : string;
+}
+
+type pty_window = {
+  row : int32;
+  col : int32;
+  xpixel : int32;
+  ypixel : int32
+}
+
+external open_pty : unit -> pty = "eio_unix_open_pty"
+external set_window_size : pty -> pty_window -> unit = "eio_unix_window_size"
+external tty_window_size : unit -> pty_window = "eio_unix_tty_window_size"

lib/mirage-crypto/rng/eio/mirage_crypto_rng_eio.ml

@@ -46,7 +46,7 @@ let running = ref false
 
 let run
     ?g
-    ?(sleep = Duration.of_sec 1)
+    ?(sleep = Duration.of_sec 100)
     generator
     env
     fn

src/ark/api/client.ml

@@ -94,13 +94,14 @@ module Agent = struct
     Capability.call_for_value t method_id request
     |> Result.map Results.exit_code_get
 
-  let spawn cmd pout t =
+  let spawn ~pty cmd pout t =
     let open Raw.Client.Agent.Spawn in
     let request, params = Capability.Request.create Params.init_pointer in
     let cmd_params = Params.cmd_init params in
     let binary, args = cmd in
     Raw.Builder.Command.binary_set cmd_params binary;
     let _ = Raw.Builder.Command.args_set_array cmd_params args in
     Params.pout_set params (Some pout);
+    Params.pty_set params pty;
     Capability.call_for_caps t method_id request Results.pin_get_pipelined
 end

src/ark/api/client.mli

@@ -34,7 +34,7 @@ module Agent : sig
   val exec :
     cmd:command -> t -> (int32, [> `Capnp of Capnp_rpc.Error.t ]) result
 
-  val spawn : command -> Process.Out.t -> t -> Process.In.t
+  val spawn : pty:bool -> command -> Process.Out.t -> t -> Process.In.t
 end
 
 module ClusterMember : sig

src/ark/api/cluster.capnp

@@ -24,7 +24,7 @@ struct HostInfo {
 
 interface Agent {
   exec @0 (cmd :Command) -> (exitCode :Int32);
-  spawn @1 (cmd: Command, pout: ProcessOut) -> (pin: ProcessIn);
+  spawn @1 (cmd: Command, pout: ProcessOut, pty: Bool) -> (pin: ProcessIn);
 }
 
 interface ClusterMember {

src/ark/bin/ark_agent.ml

@@ -26,10 +26,8 @@ let reporter =
   in
   { Logs.report }
 
-let run_client ~mgr cluster =
+let run_client ~mgr cluster hostname =
   Eio.Switch.run @@ fun sw ->
-  Eio.Switch.on_release sw (fun () -> Eio.traceln "X2 finished");
-  let hostname = "alpha" in
   Capability.with_ref (Ark.Server.agent ~sw mgr) @@ fun callback ->
   match Ark.Agents.hostinfo () with
   | Error (`Msg msg) ->
@@ -41,26 +39,29 @@ let run_client ~mgr cluster =
       (* TODO register sig handler for unregister *)
       Eio.Fiber.await_cancel ()
 
-let connect uri =
+let connect uri hostname =
   Eio_main.run @@ fun env ->
   Mirage_crypto_rng_eio.run (module Mirage_crypto_rng.Fortuna) env @@ fun () ->
   Eio.Switch.run @@ fun sw ->
-  Eio.Switch.on_release sw (fun () -> Eio.traceln "X1 finished");
   Eio.traceln "Connecting to cluster service at: %a@." Uri.pp_hum uri;
   let client_vat = Capnp_rpc_unix.client_only_vat ~sw (Eio.Stdenv.net env) in
   let sr = Capnp_rpc_unix.Vat.import_exn client_vat uri in
   let proxy_to_service = Sturdy_ref.connect_exn sr in
-  run_client ~mgr:(Eio.Stdenv.process_mgr env) proxy_to_service
+  run_client ~mgr:(Eio.Stdenv.process_mgr env) proxy_to_service hostname
 
 open Cmdliner
 
+let hostname =
+  let i = Arg.info [] ~docv:"HOSTNAME" ~doc:"Hostname of this node to register with cluster" in
+  Arg.(required @@ pos 1 (some string) None i)
+
 let connect_addr =
   let i = Arg.info [] ~docv:"ADDR" ~doc:"Address of server (capnp://...)" in
   Arg.(required @@ pos 0 (some Capnp_rpc_unix.sturdy_uri) None i)
 
 let connect_cmd =
   Cmd.v (Cmd.info "agent" ~doc:"run the agent")
-    Term.(const connect $ connect_addr)
+    Term.(const connect $ connect_addr $ hostname)
 
 let () =
   Fmt_tty.setup_std_outputs ();

src/ark/bin/ark_server.ml

@@ -26,9 +26,8 @@ let reporter =
   in
   { Logs.report }
 
-let clmember_cap_file = "cluster_member.cap"
-
-let cluser_cap_file = "cluster_user.cap"
+let clmember_cap_file = "ark_agent.cap"
+let cluser_cap_file = "ark_shell.cap"
 
 let write_cap vat sid file =
   match Capnp_rpc_unix.Cap_file.save_service vat sid file with

src/ark/bin/ark_shell.ml

@@ -26,51 +26,48 @@ let reporter =
   in
   { Logs.report }
 
-let run_client ~sw cluster stdin =
-  let hostname = "alpha" in
+let run_client hostname pty cluster env =
   let cmd = ("/usr/bin/bash", [| "bash";  |]) in
   Capability.with_ref cluster @@ fun t ->
   let agent = Client.ClusterUser.find ~hostname t in
   let stdout_q = Eio.Stream.create 100 in
   let stderr_q = Eio.Stream.create 100 in
   let on_complete_t, on_complete_u = Eio.Promise.create () in
   let pout = Ark.Server.process_out stdout_q stderr_q on_complete_u in
-  let pin = Client.Agent.spawn cmd pout agent in
-  let print_stream q =
-   while true do
-     Eio.Stream.take q |> Cstruct.to_string |> print_endline
-   done
-  in
-  Eio.Fiber.fork ~sw (fun () -> print_stream stdout_q);
-  Eio.Fiber.fork ~sw (fun () -> print_stream stderr_q);
-  Eio.Flow.copy stdin (Ark.Server.Eiox.capnp_sink (fun chunk -> Ark_api.Client.Process.In.stdin ~chunk pin));
-  let exit_code = Eio.Promise.await on_complete_t in
-  (* TODO flush stdout *)
-  Logs.info (fun l -> l "exit code %ld" exit_code);
-  (* TODO triggers shutdown trace *)
-  (* exit (Int32.to_int exit_code) *)
-  Eio.Fiber.await_cancel ()
+  let pin = Client.Agent.spawn ~pty cmd pout agent in
+  let stdin = Ark.Eiox.capnp_sink (fun chunk -> Ark_api.Client.Process.In.stdin ~chunk pin) in
+  let stdout = Ark.Eiox.stream_source stdout_q in
+  let pty_fn = if pty then Ark.Eiox.run_with_raw_term else Ark.Eiox.run in
+  pty_fn env ~stdin ~stdout on_complete_t
 
-let connect uri =
+let connect uri hostname pty =
   Eio_main.run @@ fun env ->
   Mirage_crypto_rng_eio.run (module Mirage_crypto_rng.Fortuna) env @@ fun () ->
   Eio.Switch.run @@ fun sw ->
   Logs.info (fun l -> l "Connecting to cluster service at: %a" Uri.pp_hum uri);
   let client_vat = Capnp_rpc_unix.client_only_vat ~sw (Eio.Stdenv.net env) in
   let sr = Capnp_rpc_unix.Vat.import_exn client_vat uri in
   let proxy_to_service = Sturdy_ref.connect_exn sr in
-  Eio.Switch.run @@ fun sw ->
-  run_client ~sw proxy_to_service (env#stdin)
+  let exit_code = run_client hostname pty proxy_to_service env in
+  exit (Int32.to_int exit_code)
 
 open Cmdliner
 
+let pty =
+  let info = Arg.info ["t"] ~doc:"Force allocation of a pseudoterminal" in
+  Arg.value (Arg.flag info)
+
 let connect_addr =
   let i = Arg.info [] ~docv:"ADDR" ~doc:"Address of server (capnp://...)" in
   Arg.(required @@ pos 0 (some Capnp_rpc_unix.sturdy_uri) None i)
 
+let hostname =
+  let i = Arg.info [] ~docv:"HOSTNAME" ~doc:"Hostname of this node to register with cluster" in
+  Arg.(required @@ pos 1 (some string) None i)
+
 let connect_cmd =
-  let doc = "run the client" in
-  Cmd.v (Cmd.info "connect" ~doc) Term.(const connect $ connect_addr)
+  let doc = "run the remote shell" in
+  Cmd.v (Cmd.info "ark-shell" ~doc) Term.(const connect $ connect_addr $ hostname $ pty)
 
 let () =
   Fmt_tty.setup_std_outputs ();

src/ark/lib/dune

@@ -1,3 +1,3 @@
 (library
  (name ark)
- (libraries ark_api osrelease))
+ (libraries eio.unix eio_posix ark_api osrelease))