Merge pull request #2129 from samoht/fdd

sdk: add a file-descriptor sharing server

projects/miragesdk/src/fdd/.gitignore

@@ -0,0 +1,4 @@
+*.install
+.merlin
+_build
+*~

projects/miragesdk/src/fdd/Dockerfile

@@ -0,0 +1,12 @@
+FROM ocaml/opam:alpine as base
+RUN sudo apk add m4
+RUN opam install jbuilder lwt fd-send-recv logs fmt cmdliner
+ADD . /src
+RUN opam pin add fdd /src
+RUN sudo mkdir /out
+RUN sudo cp /home/opam/.opam/4.04.2/bin/fdd /out
+
+FROM scratch
+COPY --from=base /out/fdd .
+USER 0
+ENTRYPOINT ["/fdd"]

projects/miragesdk/src/fdd/Makefile

@@ -0,0 +1,7 @@
+.PHONY: all clean
+
+all:
+	jbuilder build --dev
+
+clean:
+	jbuilder clean

projects/miragesdk/src/fdd/README.md

@@ -0,0 +1,22 @@
+## fdd -- file-descriptor daemon
+
+`Fdd` allows to share socketpair over a unix domain socket. The typical flow is
+as follows:
+
+1. Start the fdd daemon:
+    ```
+    $ fdd init
+   ```
+
+2. Create a bunch of socketpair shares:
+    ```
+    $ fdd share /tmp/foo
+    $ fdd share /tmp/bar
+    ```
+
+This will create `/tmp/foo` and `/tmp/bar` that process clients can connect too.
+Once connected, they can use `recvmsg`[1] to receive each side of the
+socketpair. If two different process do this, they then have a channel to talk
+to each other.
+
+[1]: https://linux.die.net/man/2/recvmsg

projects/miragesdk/src/fdd/common.ml

@@ -0,0 +1,45 @@
+open Lwt.Infix
+
+let src = Logs.Src.create "fdd/common"
+module Log = (val Logs.src_log src : Logs.LOG)
+
+let magic_header = "FDD0"
+
+let bind path =
+  Log.debug (fun l -> l "bind %s" path);
+  Lwt.catch (fun () -> Lwt_unix.unlink path) (fun _ -> Lwt.return ())
+  >>= fun () ->
+  let fd = Lwt_unix.socket Lwt_unix.PF_UNIX Lwt_unix.SOCK_STREAM 0 in
+  Lwt_unix.bind fd (Lwt_unix.ADDR_UNIX path) >|= fun () ->
+  fd
+
+let connect path =
+  Log.debug (fun l -> l "connect %s" path);
+  let fd = Lwt_unix.socket Lwt_unix.PF_UNIX Lwt_unix.SOCK_STREAM 0 in
+  Lwt_unix.connect fd (Lwt_unix.ADDR_UNIX path) >|= fun () ->
+  fd
+
+let send_fd ~to_send fd =
+  Log.debug (fun l -> l "send_fd");
+  let fd = Lwt_unix.unix_file_descr fd in
+  let to_send = Lwt_unix.unix_file_descr to_send in
+  let len = String.length magic_header in
+  Lwt_preemptive.detach (fun () ->
+      let i = Fd_send_recv.send_fd fd magic_header 0 len [] to_send in
+      assert (i = len)
+    ) ()
+
+let recv_fd fd =
+  Log.debug (fun l -> l "recv_fd");
+  let len = String.length magic_header in
+  let buf = Bytes.create len in
+  let fd = Lwt_unix.unix_file_descr fd in
+  Lwt_preemptive.detach (fun () ->
+      Unix.clear_nonblock fd;
+      Fd_send_recv.recv_fd fd buf 0 len []
+    ) ()
+  >|= fun (n, _, c) ->
+  Log.debug (fun l -> l "recv_fd: received %S (%d)" buf n);
+  assert (n = len);
+  assert (buf = magic_header);
+  Lwt_unix.of_unix_file_descr c

projects/miragesdk/src/fdd/common.mli

@@ -0,0 +1,4 @@
+val bind: string -> Lwt_unix.file_descr Lwt.t
+val connect: string -> Lwt_unix.file_descr Lwt.t
+val send_fd: to_send:Lwt_unix.file_descr -> Lwt_unix.file_descr -> unit Lwt.t
+val recv_fd: Lwt_unix.file_descr -> Lwt_unix.file_descr Lwt.t

projects/miragesdk/src/fdd/fdd.ml

@@ -0,0 +1,71 @@
+open Cmdliner
+
+let socket =
+  let doc =
+    Arg.info ~docv:"PATH"
+      ~doc:"Socket to communicate with the FDD server." ["s"; "socket"]
+  in
+  Arg.(value & opt string "/var/run/fdd.sock" doc)
+
+let share =
+  let doc =
+    Arg.info ~docv:"PATH" ~doc:"The path to use to share the socketpair." []
+  in
+  Arg.(required & pos 0 (some string) None doc)
+
+let setup_log style_renderer level =
+  Fmt_tty.setup_std_outputs ?style_renderer ();
+  Logs.set_level level;
+  let pp_header ppf x =
+    Fmt.pf ppf "%5d: %a " (Unix.getpid ()) Logs_fmt.pp_header x
+  in
+  Logs.set_reporter (Logs_fmt.reporter ~pp_header ());
+  ()
+
+let setup_log =
+  Term.(const setup_log $ Fmt_cli.style_renderer () $ Logs_cli.level ())
+
+let run f = Lwt_main.run f
+
+let init =
+  let f socket () = run (Init.f socket) in
+  Term.(const f $ socket $ setup_log),
+  Term.info "init" ~doc:"Start the FDD server"
+
+let test =
+  let f share () = run (Test.f share) in
+  Term.(const f $ share $ setup_log),
+  Term.info "test" ~doc:"Test a socketpair share."
+
+let share =
+  let f socket share () = run (Share.f ~socket ~share) in
+  Term.(const f $ socket $ share $ setup_log),
+  Term.info "share" ~doc:"Share a new socketpair on a given unix domain socket."
+
+let default =
+  let usage () =
+    Fmt.pr "usage: fdd [--version]\n\
+           \           [--help]\n\
+           \           <command> [<args>]\n\
+            \n\
+            The most commonly used subcommands are:\n\
+           \    init        start a new FDD server\n\
+           \    share       share a new socketpair\n\
+           \    test        test a socketpair share\n\
+            \n\
+            See `fdd help <command>` for more information on a specific \
+            command.\n%!"
+  in
+  Term.(const usage $ const ()),
+  Term.info "fdd" ~version:"%%VERSION%%"
+    ~doc:"Share socketpairs over unix domain sockets."
+
+let cmds = [
+  init;
+  share;
+  test;
+]
+
+let () = match Term.eval_choice default cmds with
+  | `Error _ -> exit 1
+  | `Ok () |`Help |`Version -> exit 0

projects/miragesdk/src/fdd/fdd.opam

@@ -0,0 +1,19 @@
+opam-version: "1.2"
+maintainer:   "Thomas Gazagnaire <thomas@gazagnaire.org>"
+authors:      "Thomas Gazagnaire <thomas@gazagnaire.org>"
+homepage:     "https://github.com/linuxkit/linuxkit"
+bug-reports:  "https://github.com/linuxkit/linuxkit/issues"
+license:      "Apache"
+dev-repo:     "https://github.com/linuxkit/linuxkit.git"
+build: [
+  ["jbuilder" "subst"]{pinned}
+  ["jbuilder" "build" "-p" name "-j" jobs]
+]
+depends: [
+  "jbuilder" {build & >= "1.0+beta10"}
+  "lwt"
+  "fd-send-recv"
+  "logs"
+  "fmt"
+  "cmdliner"
+]

projects/miragesdk/src/fdd/init.ml

@@ -0,0 +1,61 @@
+open Lwt.Infix
+open Common
+
+let src = Logs.Src.create "fdd/init"
+module Log = (val Logs.src_log src : Logs.LOG)
+
+let write_pid socket =
+  let pid_file = Filename.chop_extension socket ^ ".pid" in
+  if Sys.file_exists pid_file then (
+    Fmt.pr "Cannot start, as %s already exists.\n%!" pid_file;
+    exit 1
+  );
+  Lwt_unix.openfile pid_file Lwt_unix.[O_CREAT; O_EXCL] 0o644 >>= fun fd ->
+  Log.info (fun l -> l "Writing %s" pid_file);
+  let oc = Lwt_io.of_fd ~mode:Lwt_io.Output fd in
+  Lwt_io.write_line oc (string_of_int (Unix.getpid ())) >>= fun () ->
+  Lwt_io.close oc >|= fun () ->
+  at_exit (fun () ->
+      Log.info (fun l -> l "Removing %s" pid_file);
+      Unix.unlink pid_file
+    )
+
+(* listen on fd and send the socketpair to the first 2 connections.*)
+let send_socketpair fd =
+  let f, d = Lwt_unix.(socketpair PF_UNIX SOCK_STREAM 0) in
+  let send to_send =
+    Lwt_unix.accept fd >>= fun (fd, _) ->
+    Log.info (fun l -> l "New client!");
+    send_fd ~to_send fd
+  in
+  Lwt_unix.listen fd 2;
+  Lwt.join [send f; send d]
+
+let recv_path fd =
+  let ic = Lwt_io.of_fd ~mode:Lwt_io.Input fd in
+  Lwt_io.read_line ic >>= fun line ->
+  let path = String.trim line in
+  bind path >>= fun fd ->
+  send_socketpair fd >>= fun () ->
+  Lwt_unix.unlink path
+
+let listen fd =
+  let rec loop () =
+    Lwt_unix.accept fd >>= fun (fd, _) ->
+    Log.debug (fun l -> l "New client connected!");
+    Lwt.async (fun () ->
+        Lwt.catch
+          (fun () -> recv_path fd)
+          (fun e  ->
+             Log.err (fun l -> l "asynchronous exn: %a" Fmt.exn e);
+             Lwt.return ())
+      );
+    loop ()
+  in
+  Lwt_unix.listen fd 10;
+  loop ()
+
+let f socket =
+  write_pid socket >>= fun () ->
+  bind socket >>= fun fd ->
+  listen fd

projects/miragesdk/src/fdd/init.mli

@@ -0,0 +1 @@
+val f: string -> unit Lwt.t

projects/miragesdk/src/fdd/jbuild

@@ -0,0 +1,7 @@
+(executable
+  ((name         fdd)
+    (public_name fdd)
+    (package     fdd)
+    (flags     (:standard -cclib -static))
+    (libraries (unix lwt.unix lwt.preemptive fd-send-recv logs fmt cmdliner
+                fmt.cli logs.fmt fmt.tty logs.cli))))

projects/miragesdk/src/fdd/share.ml

@@ -0,0 +1,8 @@
+open Lwt.Infix
+open Common
+
+let f ~socket ~share =
+  connect socket >>= fun fd ->
+  let oc = Lwt_io.of_fd ~mode:Lwt_io.Output fd in
+  Lwt_io.write_line oc share >>= fun () ->
+  Lwt_io.close oc

projects/miragesdk/src/fdd/share.mli

@@ -0,0 +1 @@
+val f: socket:string -> share:string -> unit Lwt.t

projects/miragesdk/src/fdd/test.ml

@@ -0,0 +1,27 @@
+open Lwt.Infix
+open Common
+
+let get_fd share = connect share >>= recv_fd
+
+let red = Fmt.(styled `Red string)
+let green = Fmt.(styled `Green string)
+
+let f share =
+  if not (Sys.file_exists share) then (
+    Fmt.pr "%a %s does not exist.\n%!" red "[ERROR]" share;
+    exit 1;
+  );
+  get_fd share >>= fun x ->
+  get_fd share >>= fun y ->
+  let x = Lwt_io.of_fd ~mode:Lwt_io.Output x in
+  let y = Lwt_io.of_fd ~mode:Lwt_io.Input y in
+  let payload = "This is a test!" in
+  Lwt_io.write_line x payload >>= fun () ->
+  Lwt_io.read_line y >|= fun buf ->
+  if buf <> payload then (
+    Fmt.pr "%a Expecting %S, but got %S.\n%!" red "[ERROR]" payload buf;
+    exit 1
+  ) else (
+    Fmt.pr "%a the socketpair which was shared on %s is working properly.\n%!"
+      green "[SUCCES]" share
+  )

projects/miragesdk/src/fdd/test.mli

@@ -0,0 +1 @@
+val f: string -> unit Lwt.t