feat(action-plugin): run dynamic-run over RPC

doc/reference/actions/dynamic-run.rst

@@ -8,6 +8,9 @@ dynamic-run
    Execute a program that was linked against the ``dune-action-plugin`` library.
    ``<prog>`` is resolved in the same way as in :doc:`run`.
 
+   The program remains running while Dune builds dependencies that it discovers,
+   so each ``dynamic-run`` invocation starts the program only once.
+
    Example::
 
    (dynamic-run ./plugin.exe)

otherlibs/dune-action-plugin/src/dune

@@ -1,6 +1,6 @@
 (library
  (name dune_action_plugin)
  (public_name dune-action-plugin)
- (libraries stdune csexp dune-glob unix dune-rpc)
+ (libraries stdune csexp dune-glob unix threads.posix dune-rpc)
  (synopsis
   "[Internal] Monadic interface for defining scripts with dynamic or complex sets of dependencies."))

otherlibs/dune-action-plugin/src/dune_action_plugin.ml

@@ -134,11 +134,305 @@ module V1 = struct
       }
   ;;
 
+  module Dap_client = struct
+    module Rpc = Protocol.Rpc
+    module Dune_rpc = Dune_rpc.V1
+
+    module Error = struct
+      type t =
+        | Version_error of
+            { procedure : string
+            ; error : Dune_rpc.Version_error.t
+            }
+        | Response_error of Dune_rpc.Response.Error.t
+        | Build_error of string
+
+      let message = function
+        | Version_error { procedure; error } ->
+          "unable to negotiate " ^ procedure ^ ": " ^ Dune_rpc.Version_error.message error
+        | Response_error error ->
+          "dune rpc error: " ^ Dune_rpc.Response.Error.message error
+        | Build_error message -> message
+      ;;
+    end
+
+    module type Monad = sig
+      type 'a t
+
+      val return : 'a -> 'a t
+
+      module O : sig
+        val ( let* ) : 'a t -> ('a -> 'b t) -> 'b t
+        val ( let+ ) : 'a t -> ('a -> 'b) -> 'b t
+      end
+    end
+
+    module Make
+        (M : Monad)
+        (Chan : sig
+           type t
+         end)
+        (Client : Dune_rpc.Client.S with type 'a fiber := 'a M.t and type chan := Chan.t) =
+    struct
+      type t =
+        { client : Client.t
+        ; action_id : string
+        ; targets : String.Set.t
+        ; build_deps_request : (Rpc.Build_deps.t, string option) Client.Versioned.request
+        ; mutable prepared_dependencies : Dependency.Set.t
+        }
+
+      let targets t = t.targets
+      let prepared_dependencies t = t.prepared_dependencies
+      let version_error procedure error = Error.Version_error { procedure; error }
+      let response_error error = Error.Response_error error
+
+      let prepare_request client name request =
+        let open M.O in
+        let+ result = Client.Versioned.prepare_request client request in
+        Stdune.Result.map_error result ~f:(fun error -> version_error name error)
+      ;;
+
+      let request client request payload =
+        let open M.O in
+        let+ result = Client.request client request payload in
+        Stdune.Result.map_error result ~f:response_error
+      ;;
+
+      let create client ~action_id =
+        let open M.O in
+        let* initialize_request =
+          prepare_request client "dap/initialize" Rpc.initialize_request
+        in
+        match initialize_request with
+        | Error _ as error -> M.return error
+        | Ok initialize_request ->
+          let* build_deps_request =
+            prepare_request client "dap/build-deps" Rpc.build_deps_request
+          in
+          (match build_deps_request with
+           | Error _ as error -> M.return error
+           | Ok build_deps_request ->
+             let* run_arguments = request client initialize_request action_id in
+             (match run_arguments with
+              | Error _ as error -> M.return error
+              | Ok run_arguments ->
+                M.return
+                  (Ok
+                     { client
+                     ; action_id
+                     ; targets = run_arguments.targets
+                     ; build_deps_request
+                     ; prepared_dependencies = Dependency.Set.empty
+                     })))
+      ;;
+
+      let connect chan ~action_id ~f =
+        let open M.O in
+        let id = Dune_rpc.Request.Id.make (List [ Atom "dap"; Atom action_id ]) in
+        let initialize = Dune_rpc.Request.Initialize.create ~id in
+        Client.connect chan initialize ~f:(fun client ->
+          let* t = create client ~action_id in
+          match t with
+          | Error _ as error -> M.return error
+          | Ok t -> f t)
+      ;;
+
+      let build_deps t deps =
+        if Dependency.Set.is_empty deps
+        then M.return (Ok ())
+        else
+          let open M.O in
+          let+ response =
+            request
+              t.client
+              t.build_deps_request
+              { Rpc.Build_deps.action_id = t.action_id; deps }
+          in
+          match response with
+          | Error _ as error -> error
+          | Ok (Some message) -> Error (Error.Build_error message)
+          | Ok None ->
+            t.prepared_dependencies <- Dependency.Set.union t.prepared_dependencies deps;
+            Ok ()
+      ;;
+    end
+  end
+
+  module Blocking_rpc = struct
+    module Fiber = struct
+      type 'a t = unit -> 'a
+
+      let return x () = x
+
+      module O = struct
+        let ( let* ) x f () = f (x ()) ()
+        let ( let+ ) x f () = f (x ())
+      end
+
+      let collect_errors f () =
+        match f () () with
+        | result -> Ok result
+        | exception exn -> Error [ exn ]
+      ;;
+
+      let finalize f ~finally () =
+        Exn.protect ~f:(fun () -> f () ()) ~finally:(fun () -> finally () ())
+      ;;
+
+      let parallel_iter next ~f () =
+        let rec loop () =
+          match next () () with
+          | None -> ()
+          | Some x ->
+            f x ();
+            loop ()
+        in
+        loop ()
+      ;;
+
+      module Ivar = struct
+        type 'a t =
+          { mutex : Mutex.t
+          ; condition : Condition.t
+          ; mutable value : 'a option
+          }
+
+        let create () =
+          { mutex = Mutex.create (); condition = Condition.create (); value = None }
+        ;;
+
+        let read t () =
+          Mutex.lock t.mutex;
+          let rec loop () =
+            match t.value with
+            | Some value ->
+              Mutex.unlock t.mutex;
+              value
+            | None ->
+              Condition.wait t.condition t.mutex;
+              loop ()
+          in
+          loop ()
+        ;;
+
+        let fill t value () =
+          Mutex.lock t.mutex;
+          (match t.value with
+           | Some _ -> ()
+           | None ->
+             t.value <- Some value;
+             Condition.broadcast t.condition);
+          Mutex.unlock t.mutex
+        ;;
+      end
+
+      let thread_pool =
+        lazy
+          (Stdune.Thread_pool0.create
+             ~spawn:(fun f -> Thread.create f ())
+             ~min_workers:0
+             ~max_workers:50)
+      ;;
+
+      let fork_and_join_unit f g () =
+        Stdune.Thread_pool0.task (Lazy.force thread_pool) ~f:(fun () ->
+          match f () () with
+          | () -> ()
+          | exception exn ->
+            prerr_endline ("dune rpc reader failed: " ^ Printexc.to_string exn);
+            flush stderr;
+            exit 2);
+        g () ()
+      ;;
+    end
+
+    module Chan = struct
+      type t =
+        { socket : Unix.file_descr
+        ; ic : in_channel
+        ; oc : out_channel
+        }
+
+      let read t () =
+        match Csexp.input_opt t.ic with
+        | Ok sexp -> sexp
+        | Error message -> failwith ("unable to read dune rpc packet: " ^ message)
+        | exception End_of_file -> None
+        | exception Sys_error _ -> None
+        | exception Unix.Unix_error _ -> None
+      ;;
+
+      let write t packets () =
+        List.iter packets ~f:(Csexp.to_channel t.oc);
+        flush t.oc
+      ;;
+
+      let close t () =
+        (match Unix.shutdown t.socket Unix.SHUTDOWN_ALL with
+         | () -> ()
+         | exception Unix.Unix_error _ -> ());
+        close_out_noerr t.oc;
+        close_in_noerr t.ic
+      ;;
+
+      let create socket =
+        { socket
+        ; ic = Unix.in_channel_of_descr (Unix.dup socket)
+        ; oc = Unix.out_channel_of_descr socket
+        }
+      ;;
+    end
+
+    module Client = Dune_rpc.V1.Client.Make (Fiber) (Chan)
+
+    let connect where =
+      let connection_error exn =
+        let message =
+          match exn with
+          | Unix.Unix_error (error, syscall, arg) ->
+            Unix_error.Detailed.create error ~syscall ~arg
+            |> Unix_error.Detailed.to_string_hum
+          | _ -> raise exn
+        in
+        Execution_error.raise ("unable to connect to dune rpc server: " ^ message)
+      in
+      let socket_of_addr addr =
+        let domain =
+          match addr with
+          | Unix.ADDR_UNIX _ -> Unix.PF_UNIX
+          | Unix.ADDR_INET _ -> Unix.PF_INET
+        in
+        let socket =
+          match Unix.socket domain Unix.SOCK_STREAM 0 with
+          | socket -> socket
+          | exception exn -> connection_error exn
+        in
+        match Unix.connect socket addr with
+        | () -> Chan.create socket
+        | exception exn ->
+          Unix.close socket;
+          connection_error exn
+      in
+      match where with
+      | `Unix path -> socket_of_addr (Unix.ADDR_UNIX path)
+      | `Ip (`Host host, `Port port) ->
+        let service = Int.to_string port in
+        (match Unix.getaddrinfo host service [ Unix.AI_SOCKTYPE Unix.SOCK_STREAM ] with
+         | [] -> Execution_error.raise ("unable to resolve dune rpc host: " ^ host)
+         | addr :: _ -> socket_of_addr addr.Unix.ai_addr
+         | exception exn -> connection_error exn)
+    ;;
+  end
+
+  module Blocking_dap_client =
+    Dap_client.Make (Blocking_rpc.Fiber) (Blocking_rpc.Chan) (Blocking_rpc.Client)
+
   let rec run_by_dune t context =
     match t with
-    | Pure () -> Context.respond context Done
+    | Pure () -> Ok ()
     | Stage at ->
-      let allowed_targets = Context.targets context in
+      let allowed_targets = Blocking_dap_client.targets context in
       let disallowed_targets = String.Set.diff at.targets allowed_targets in
       (match String.Set.to_list disallowed_targets with
        | [] -> ()
@@ -155,13 +449,14 @@ module V1 = struct
                dune file:\n\
                %sTo fix, add them to target list in dune file."
               (ts |> String.concat ~sep:"\n")));
-      let prepared_dependencies = Context.prepared_dependencies context in
       let required_dependencies =
-        Dependency.Set.diff at.dependencies prepared_dependencies
+        Dependency.Set.diff
+          at.dependencies
+          (Blocking_dap_client.prepared_dependencies context)
       in
-      if Dependency.Set.is_empty required_dependencies
-      then run_by_dune (at.action ()) context
-      else Context.respond context (Need_more_deps required_dependencies)
+      (match Blocking_dap_client.build_deps context required_dependencies () with
+       | Error _ as error -> error
+       | Ok () -> run_by_dune (at.action ()) context)
   ;;
 
   (* If executable is not run by dune, assume that all dependencies are already
@@ -173,15 +468,33 @@ module V1 = struct
   ;;
 
   let do_run t =
-    match Protocol.Context.create () with
-    | Run_outside_of_dune -> run_outside_of_dune t
-    | Error message ->
+    match
+      ( Env.get Env.initial Protocol.Rpc.action_id_env_variable
+      , Env.get Env.initial Protocol.old_run_by_dune_env_variable )
+    with
+    | None, None -> run_outside_of_dune t
+    | None, Some _ ->
       Execution_error.raise
-        (Printf.sprintf
-           "Error during communication with dune. %s Did you use different dune version \
-            to compile the executable?"
-           message)
-    | Ok context -> run_by_dune t context
+        "this dune-action-plugin executable requires Dune's RPC dynamic-run protocol"
+    | Some action_id, _ ->
+      let where =
+        match Dune_rpc.Private.Where.of_env Env.initial with
+        | Ok where -> where
+        | Error `Missing -> Execution_error.raise "unable to find a dune rpc server"
+        | Error (`Exn exn) ->
+          Execution_error.raise
+            ("invalid dune rpc server address: " ^ Printexc.to_string exn)
+      in
+      let chan = Blocking_rpc.connect where in
+      (match
+         Blocking_dap_client.connect
+           chan
+           ~action_id
+           ~f:(fun context () -> run_by_dune t context)
+           ()
+       with
+       | Ok () -> ()
+       | Error error -> Execution_error.raise (Dap_client.Error.message error))
   ;;
 
   let run t =
@@ -201,6 +514,7 @@ module V1 = struct
 
   module Private = struct
     module Protocol = Protocol
+    module Dap_client = Dap_client
 
     let do_run = do_run