Bistro_app.simple: simplified workflow execution

Provides a function for the common case where we want to run a
workflow and put some of the targets in given locations. The function
wraps the creation of a database and a scheduler, as well as
concurrency aspects.

This function is defined in the new Bistro_app module, itself inside
the new bistro.utils library.

.merlin

@@ -1,10 +1,12 @@
 B _build/lib
-B _build/std
-B _build/engine
+B _build/lib/bioinfo
+B _build/lib/engine
+B _build/lib/utils
 B _build/ppx
-S src/lib
-S src/std
-S src/engine
+S lib
+S lib/bioinfo
+S lib/engine
+S lib/utils
 S src/ppx
 S src/test
 PKG bistro

_oasis

@@ -64,6 +64,15 @@ Library bistro_engine
                     netstring, pvem, shell, threads
   XMETADescription: FIXME
 
+Library bistro_utils
+  Path:             lib/utils
+  Modules:          Bistro_app
+  FindlibName:      utils
+  FindlibParent:    bistro
+  BuildDepends:     bistro.engine
+  XMETARequires:    bistro.engine
+  XMETADescription: FIXME
+
 Executable "ppx_bistro"
   Path:             ppx
   MainIs:           ppx_bistro.ml
@@ -91,7 +100,7 @@ Executable diamond
 Executable sacCer2_pho4
   Path:             examples
   MainIs:           sacCer2_pho4.ml
-  BuildDepends:     bistro.engine, bistro.bioinfo
+  BuildDepends:     bistro.utils, bistro.bioinfo
   CompiledObject:   byte
 
 Executable rnaseq

examples/sacCer2_pho4.ml

@@ -19,24 +19,9 @@ let chIP_pho4_noPi_bam = Samtools.(indexed_bam_of_sam chIP_pho4_noPi_sam / index
 
 let chIP_pho4_noPi_macs2 = Macs2.callpeak chIP_pho4_noPi_bam
 
-
-
-
-open Bistro_engine
-open Lwt
-
-let db = Db.init_exn "_bistro"
-let scheduler = Scheduler.make ~np:4 ~mem:(10 * 1024) db
-
-
-let main () =
-  Scheduler.build scheduler chIP_pho4_noPi_macs2 >|= function
-  | `Ok p -> print_endline p
-  | `Error xs ->
-    fprintf stderr "Some workflow(s) failed:\n" ;
-    List.iter xs ~f:(fun (u, msg) ->
-        fprintf stderr "\t%s\t%s\n" (Bistro.Workflow.id' u) msg
-      ) ;
-    List.iter xs ~f:(fun (u, _) -> Db.output_report db u stderr)
-
-let () = Lwt_unix.run (main ())
+let () =
+  Bistro_app.(
+    simple ~np:4 ~mem:(10 * 1024) [
+      [ "output" ; "chIP_pho4_noPi_macs2.peaks" ] %> chIP_pho4_noPi_macs2
+    ]
+  )

lib/utils/bistro_app.ml

@@ -0,0 +1,54 @@
+open Core.Std
+open Bistro.Std
+open Bistro_engine
+open Lwt
+
+type target = string list * Bistro.Workflow.u
+
+let ( %> ) path w = path, Bistro.Workflow.u w
+
+let rec string_of_path = function
+  | [] -> "."
+  | "" :: t -> Filename.concat "." (string_of_path t)
+  | p -> List.reduce_exn p ~f:Filename.concat
+
+let make_absolute p =
+  if Filename.is_absolute p then p
+  else Filename.concat (Sys.getcwd ()) p
+
+let link p p_u =
+  let dst = string_of_path p in
+  let src = make_absolute p_u in
+  Unix.mkdir_p (Filename.dirname dst) ;
+  let cmd = sprintf "rm -rf %s && ln -s %s %s" dst src dst in
+  ignore (Sys.command cmd)
+
+let foreach_target db scheduler (dest, u) =
+  Scheduler.build' scheduler u >|= function
+  | `Ok cache_path ->
+    link dest cache_path ;
+    `Ok ()
+  | `Error xs ->
+    `Error (
+      List.map xs ~f:(fun (u, msg) ->
+          Bistro.Workflow.id' u,
+          msg,
+          Db.report db u
+        )
+    )
+
+let error_report = function
+  | `Ok () -> ()
+  | `Error xs ->
+    List.iter xs ~f:(fun (_,_,report) ->
+        prerr_endline report
+      )
+
+let simple ?(np = 1) ?(mem = 1024) targets =
+  let db = Db.init_exn "_bistro" in
+  let scheduler = Scheduler.make ~np ~mem db in
+  let results =
+    Lwt_list.map_p (foreach_target db scheduler) targets
+    |> Lwt_unix.run
+  in
+  List.iter results ~f:error_report

lib/utils/bistro_app.mli

@@ -0,0 +1,10 @@
+open Bistro.Std
+
+type target
+
+val ( %> ) : string list -> _ workflow -> target
+
+val simple :
+  ?np:int ->
+  ?mem:int ->
+  target list -> unit