feature: concurrency action

We add a (concurrent ) action which acts like (progn ) the difference
being the actions contained within can be executed concurrently by Dune.

<!-- ps-id: 28962076-9451-4253-be23-14d44a88eec0 -->

Signed-off-by: Ali Caglayan <alizter@gmail.com>

Author: Alizter

CHANGES.md

@@ -28,6 +28,9 @@ Unreleased
 - Bytecode executables built for JSOO are linked with `-noautolink` and no
   longer depend on the shared stubs of their dependent libraries (#7156, @nojb)
 
+- Added a new user action `(concurrent )` which is like `(progn )` but runs the
+  actions concurrently. (#6933, @Alizter)
+
 3.7.0 (2023-02-17)
 ------------------
 

doc/concepts.rst

@@ -789,6 +789,10 @@ The following constructions are available:
   ``setenv``, ``ignore-<outputs>``, ``with-stdin-from`` and
   ``with-<outputs>-to``. This action is available since Dune 2.0.
 - ``(progn <DSL>...)`` to execute several commands in sequence
+- ``(concurrent <DSL>...)``` to execute several commands concurrently.
+  **Warning:** This is limited by the number of available jobs to Dune.
+  Therefore care must be taken when writing actions that require a concurrent
+  run, such as running a server-client test with `-j 1`.
 - ``(echo <string>)`` to output a string on stdout
 - ``(write-file <file> <string>)`` writes ``<string>`` to ``<file>``
 - ``(cat <file> ...)`` to sequentially print the contents of files to stdout

src/dune_engine/action.ml

@@ -61,6 +61,7 @@ struct
     | Ignore (outputs, r) ->
       List [ atom (sprintf "ignore-%s" (Outputs.to_string outputs)); encode r ]
     | Progn l -> List (atom "progn" :: List.map l ~f:encode)
+    | Concurrent l -> List (atom "concurrent" :: List.map l ~f:encode)
     | Echo xs -> List (atom "echo" :: List.map xs ~f:string)
     | Cat xs -> List (atom "cat" :: List.map xs ~f:path)
     | Copy (x, y) -> List [ atom "copy"; path x; target y ]
@@ -118,6 +119,8 @@ struct
 
   let progn ts = Progn ts
 
+  let concurrent ts = Concurrent ts
+
   let echo s = Echo s
 
   let cat ps = Cat ps
@@ -289,7 +292,7 @@ let fold_one_step t ~init:acc ~f =
   | Redirect_in (_, _, t)
   | Ignore (_, t)
   | With_accepted_exit_codes (_, t) -> f acc t
-  | Progn l | Pipe (_, l) -> List.fold_left l ~init:acc ~f
+  | Progn l | Pipe (_, l) | Concurrent l -> List.fold_left l ~init:acc ~f
   | Run _
   | Dynamic_run _
   | Echo _
@@ -337,7 +340,7 @@ let rec is_dynamic = function
   | Redirect_in (_, _, t)
   | Ignore (_, t)
   | With_accepted_exit_codes (_, t) -> is_dynamic t
-  | Progn l | Pipe (_, l) -> List.exists l ~f:is_dynamic
+  | Progn l | Pipe (_, l) | Concurrent l -> List.exists l ~f:is_dynamic
   | Run _
   | System _
   | Bash _
@@ -386,7 +389,7 @@ let is_useful_to distribute memoize =
     | Redirect_out (_, _, _, t) -> memoize || loop t
     | Redirect_in (_, _, t) -> loop t
     | Ignore (_, t) | With_accepted_exit_codes (_, t) -> loop t
-    | Progn l | Pipe (_, l) -> List.exists l ~f:loop
+    | Progn l | Pipe (_, l) | Concurrent l -> List.exists l ~f:loop
     | Echo _ -> false
     | Cat _ -> memoize
     | Copy _ -> memoize

src/dune_engine/action_exec.ml

@@ -67,6 +67,15 @@ type done_or_more_deps =
      subdirectories that contains targets having the same name. *)
   | Need_more_deps of (DAP.Dependency.Set.t * Dynamic_dep.Set.t)
 
+let done_or_more_deps_union x y =
+  match (x, y) with
+  | Done, Done -> Done
+  | Done, Need_more_deps x | Need_more_deps x, Done -> Need_more_deps x
+  | Need_more_deps (deps1, dyn_deps1), Need_more_deps (deps2, dyn_deps2) ->
+    Need_more_deps
+      ( DAP.Dependency.Set.union deps1 deps2
+      , Dynamic_dep.Set.union dyn_deps1 dyn_deps2 )
+
 type exec_context =
   { targets : Targets.Validated.t option
   ; context : Build_context.t option
@@ -273,6 +282,9 @@ let rec exec t ~display ~ectx ~eenv =
   | Ignore (outputs, t) ->
     redirect_out t ~display ~ectx ~eenv ~perm:Normal outputs Config.dev_null
   | Progn ts -> exec_list ts ~display ~ectx ~eenv
+  | Concurrent ts ->
+    Fiber.parallel_map ts ~f:(exec ~display ~ectx ~eenv)
+    >>| List.fold_left ~f:done_or_more_deps_union ~init:Done
   | Echo strs ->
     let+ () = exec_echo eenv.stdout_to (String.concat strs ~sep:" ") in
     Done

src/dune_engine/action_intf.ml

@@ -41,6 +41,7 @@ module type Ast = sig
     | Redirect_in of Inputs.t * path * t
     | Ignore of Outputs.t * t
     | Progn of t list
+    | Concurrent of t list
     | Echo of string list
     | Cat of path list
     | Copy of path * target
@@ -91,6 +92,8 @@ module type Helpers = sig
 
   val progn : t list -> t
 
+  val concurrent : t list -> t
+
   val echo : string list -> t
 
   val cat : path list -> t

src/dune_engine/action_mapper.ml

@@ -30,6 +30,7 @@ module Make (Src : Action_intf.Ast) (Dst : Action_intf.Ast) = struct
       Redirect_in (inputs, f_path ~dir fn, f t ~dir)
     | Ignore (outputs, t) -> Ignore (outputs, f t ~dir)
     | Progn l -> Progn (List.map l ~f:(fun t -> f t ~dir))
+    | Concurrent l -> Concurrent (List.map l ~f:(fun t -> f t ~dir))
     | Echo xs -> Echo (List.map xs ~f:(f_string ~dir))
     | Cat xs -> Cat (List.map xs ~f:(f_path ~dir))
     | Copy (x, y) -> Copy (f_path ~dir x, f_target ~dir y)

src/dune_engine/action_to_sh.ml

@@ -15,6 +15,7 @@ module Simplified = struct
     | Redirect_in of t list * Inputs.t * source
     | Pipe of t list list * Outputs.t
     | Sh of string
+    | Concurrent of t list list
 end
 
 open Simplified
@@ -55,6 +56,7 @@ let simplify act =
     | Ignore (outputs, act) ->
       Redirect_out (block act, outputs, Dev_null) :: acc
     | Progn l -> List.fold_left l ~init:acc ~f:(fun acc act -> loop act acc)
+    | Concurrent l -> Concurrent (List.map ~f:block l) :: acc
     | Echo xs -> echo (String.concat xs ~sep:"")
     | Cat x -> cat x :: acc
     | Copy (x, y) -> Run ("cp", [ x; y ]) :: acc
@@ -172,6 +174,27 @@ and pp = function
            ; Pp.concat ~sep:(Pp.verbatim " | ") (List.map l ~f:block)
            ; Pp.verbatim end_
            ]))
+  | Concurrent t -> (
+    match t with
+    | [] -> Pp.verbatim "true"
+    | [ x ] -> block x
+    | x :: l ->
+      Pp.hovbox ~indent:2
+        (Pp.concat
+           [ Pp.char '('
+           ; Pp.space
+           ; block x
+           ; Pp.space
+           ; Pp.char '&'
+           ; Pp.space
+           ; Pp.concat ~sep:(Pp.verbatim "&") (List.map l ~f:block)
+           ; Pp.space
+           ; Pp.char '&'
+           ; Pp.space
+           ; Pp.verbatim "wait"
+           ; Pp.space
+           ; Pp.verbatim ")"
+           ]))
 
 let rec pp_seq = function
   | [] -> Pp.verbatim "true"

src/dune_lang/action.ml

@@ -85,6 +85,7 @@ type t =
   | Redirect_in of Inputs.t * String_with_vars.t * t
   | Ignore of Outputs.t * t
   | Progn of t list
+  | Concurrent of t list
   | Echo of String_with_vars.t list
   | Cat of String_with_vars.t list
   | Copy of String_with_vars.t * String_with_vars.t
@@ -210,6 +211,9 @@ let decode =
         ; ("ignore-stderr", t >>| fun t -> Ignore (Stderr, t))
         ; ("ignore-outputs", t >>| fun t -> Ignore (Outputs, t))
         ; ("progn", repeat t >>| fun l -> Progn l)
+        ; ( "concurrent"
+          , Syntax.since Stanza.syntax (3, 8) >>> repeat t >>| fun l ->
+            Concurrent l )
         ; ( "echo"
           , let+ x = sw
             and+ xs = repeat sw in
@@ -299,6 +303,7 @@ let rec encode =
   | Ignore (outputs, r) ->
     List [ atom (sprintf "ignore-%s" (Outputs.to_string outputs)); encode r ]
   | Progn l -> List (atom "progn" :: List.map l ~f:encode)
+  | Concurrent l -> List (atom "concurrent" :: List.map l ~f:encode)
   | Echo xs -> List (atom "echo" :: List.map xs ~f:sw)
   | Cat xs -> List (atom "cat" :: List.map xs ~f:sw)
   | Copy (x, y) -> List [ atom "copy"; sw x; sw y ]
@@ -355,7 +360,7 @@ let ensure_at_most_one_dynamic_run ~loc action =
     | Mkdir _
     | Diff _
     | Cram _ -> false
-    | Pipe (_, ts) | Progn ts ->
+    | Pipe (_, ts) | Progn ts | Concurrent ts ->
       List.fold_left ts ~init:false ~f:(fun acc t ->
           let have_dyn = loop t in
           if acc && have_dyn then
@@ -383,6 +388,7 @@ let rec map_string_with_vars t ~f =
   | Redirect_in (i, sw, t) -> Redirect_in (i, f sw, t)
   | Ignore (o, t) -> Ignore (o, map_string_with_vars t ~f)
   | Progn xs -> Progn (List.map xs ~f:(map_string_with_vars ~f))
+  | Concurrent xs -> Concurrent (List.map xs ~f:(map_string_with_vars ~f))
   | Echo xs -> Echo xs
   | Cat xs -> Cat (List.map ~f xs)
   | Copy (sw1, sw2) -> Copy (f sw1, f sw2)

src/dune_lang/action.mli

@@ -78,6 +78,7 @@ type t =
   | Redirect_in of Inputs.t * String_with_vars.t * t
   | Ignore of Outputs.t * t
   | Progn of t list
+  | Concurrent of t list
   | Echo of String_with_vars.t list
   | Cat of String_with_vars.t list
   | Copy of String_with_vars.t * String_with_vars.t

src/dune_rules/action_unexpanded.ml

@@ -417,6 +417,9 @@ let rec expand (t : Dune_lang.Action.t) ~context : Action.t Action_expander.t =
   | Progn l ->
     let+ l = A.all (List.map l ~f:expand) in
     O.Progn l
+  | Concurrent l ->
+    let+ l = A.all (List.map l ~f:expand) in
+    O.Concurrent l
   | Echo xs ->
     let+ l = A.all (List.map xs ~f:E.strings) in
     let l = List.concat l in