X-Git-Url: http://git.annexia.org/?a=blobdiff_plain;f=daemon%2Fdaemon.ml;h=c04bfcbac06cce362b20ac81cace23d6c30ae2e0;hb=77707acb12b5424488757569f376e1d9b58e5a22;hp=25b81b7a86a4caa7e8b3ab51edd59e42cc647d5f;hpb=0c9faf57f9239b0fe1c0b46353d222bb4cf5cd74;p=whenjobs.git

diff --git a/daemon/daemon.ml b/daemon/daemon.ml
index 25b81b7..c04bfcb 100644
--- a/daemon/daemon.ml
+++ b/daemon/daemon.ml
@@ -78,6 +78,12 @@ let rec init j d =
       ~proc_get_jobs
       ~proc_cancel_job
       ~proc_start_job
+      ~proc_get_job
+      ~proc_set_variables
+      ~proc_get_job_names
+      ~proc_test_variables
+      ~proc_ping_daemon
+      ~proc_whisper_variables
       (Rpc_server.Unix addr)
       Rpc.Tcp (* not TCP, this is the same as SOCK_STREAM *)
       Rpc.Socket
@@ -93,7 +99,7 @@ let rec init j d =
 and proc_reload_file () =
   if !debug then Syslog.notice "remote call: reload_file";
 
-  try reload_file (); `ok
+  try reload_files (); `ok
   with Failure err -> `error err
 
 and proc_set_variable (name, value) =
@@ -106,8 +112,10 @@ and proc_set_variable (name, value) =
     state := Whenstate.set_variable !state name value;
 
     (* Which jobs need to be re-evaluated? *)
-    let jobs = Whenstate.get_dependencies !state name in
-    reevaluate_whenjobs jobs;
+    let jobs = Whenstate.get_dependencies !state [name] in
+    let jobnames, state' = reevaluate_whenjobs !state jobs in
+    let state' = run_whenjobs state' jobnames in
+    state := state';
 
     `ok
   with
@@ -161,15 +169,139 @@ and proc_cancel_job serial =
 and proc_start_job jobname =
   try
     let job = Whenstate.get_job !state jobname in
-    run_job job;
+    let state' = run_job !state job in
+    state := state';
     `ok
   with
   | Not_found -> `error "job not found"
   | exn -> `error (Printexc.to_string exn)
 
-(* Reload the jobs file. *)
-and reload_file () =
-  let file = sprintf "%s/jobs.cmo" !jobsdir in
+and proc_get_job serial =
+  try
+    let serial = big_int_of_string serial in
+    let pid = BigIntMap.find serial !serialmap in
+    let job, dir, serial, start_time = IntMap.find pid !runningmap in
+    { Whenproto_aux.job_name = job.job_name;
+      job_serial = string_of_big_int serial;
+      job_tmpdir = dir; job_start_time = Int64.of_float start_time }
+  with
+  | Not_found -> failwith "job not found"
+  | exn -> failwith (Printexc.to_string exn)
+
+and proc_set_variables vars =
+  try
+    let vars = Array.map (
+      fun { Whenproto_aux.sv_name = name; sv_value = value } ->
+        name, variable_of_rpc value
+    ) vars in
+    let vars = Array.to_list vars in
+
+    if !debug then
+      Syslog.notice "remote call: set_variables (%s)"
+        (String.concat " "
+           (List.map (
+             fun (name, value) ->
+               sprintf "%s=%s" name (string_of_variable value)
+            ) vars));
+
+    List.iter (fun (name, _) -> check_valid_variable_name name) vars;
+
+    (* Update all the variables atomically. *)
+    let s = List.fold_left (
+      fun s (name, value) -> Whenstate.set_variable s name value
+    ) !state vars in
+    state := s;
+
+    (* Which jobs need to be re-evaluated? *)
+    let jobs = Whenstate.get_dependencies !state (List.map fst vars) in
+    let jobnames, state' = reevaluate_whenjobs !state jobs in
+    let state' = run_whenjobs state' jobnames in
+    state := state';
+
+    `ok
+  with
+    Failure msg -> `error msg
+
+and proc_get_job_names () =
+  Array.of_list (Whenstate.get_job_names !state)
+
+and proc_test_variables vars =
+  (* This is the same as proc_set_variables, except that it doesn't
+   * update the state, it just returns the jobs that *would* run if
+   * these variables were set to these values.
+   *)
+  let vars = Array.map (
+    fun { Whenproto_aux.sv_name = name; sv_value = value } ->
+      name, variable_of_rpc value
+  ) vars in
+  let vars = Array.to_list vars in
+
+  if !debug then
+    Syslog.notice "remote call: test_variables (%s)"
+      (String.concat " "
+         (List.map (
+           fun (name, value) ->
+             sprintf "%s=%s" name (string_of_variable value)
+          ) vars));
+
+  List.iter (fun (name, _) -> check_valid_variable_name name) vars;
+
+  (* Update all the variables atomically. *)
+  let state = List.fold_left (
+    fun s (name, value) -> Whenstate.set_variable s name value
+  ) !state vars in
+
+  (* Which jobs WOULD be re-evaluated? *)
+  let jobs = Whenstate.get_dependencies state (List.map fst vars) in
+  let jobnames, _ = reevaluate_whenjobs state jobs in
+
+  (* Return the names. *)
+  Array.of_list jobnames
+
+and proc_ping_daemon () = `ok
+
+and proc_whisper_variables vars =
+  try
+    let vars = Array.map (
+      fun { Whenproto_aux.sv_name = name; sv_value = value } ->
+        name, variable_of_rpc value
+    ) vars in
+    let vars = Array.to_list vars in
+
+    if !debug then
+      Syslog.notice "remote call: whisper_variables (%s)"
+        (String.concat " "
+           (List.map (
+             fun (name, value) ->
+               sprintf "%s=%s" name (string_of_variable value)
+            ) vars));
+
+    List.iter (fun (name, _) -> check_valid_variable_name name) vars;
+
+    (* Update all the variables atomically. *)
+    let s = List.fold_left (
+      fun s (name, value) -> Whenstate.set_variable s name value
+    ) !state vars in
+    state := s;
+
+    (* .. but don't reevaluate or run jobs. *)
+
+    `ok
+  with
+    Failure msg -> `error msg
+
+(* Reload the jobs file(s). *)
+and reload_files () =
+  (* Get dir/*.cmo *)
+  let dir = !jobsdir in
+  let files = Array.to_list (Sys.readdir dir) in
+  let files = List.filter (
+    fun file ->
+      let n = String.length file in
+      n >= 5 && String.sub file (n-4) 4 = ".cmo"
+  ) files in
+  let files = List.map (fun file -> dir // file) files in
+  let files = List.sort compare files in
 
   (* As we are reloading the file, we want to create a new state
    * that has no jobs, but has all the variables from the previous
@@ -180,9 +312,10 @@ and reload_file () =
 
   let s =
     try
-      Dynlink.loadfile file;
+      List.iter Dynlink.loadfile files;
       let s = Whenfile.get_state () in
-      Syslog.notice "loaded %d job(s) from %s" (Whenstate.nr_jobs s) file;
+      Syslog.notice "loaded %d job(s) from %d file(s)"
+        (Whenstate.nr_jobs s) (List.length files);
       s
     with
     | Dynlink.Error err ->
@@ -192,49 +325,56 @@ and reload_file () =
     | exn ->
       failwith (Printexc.to_string exn) in
 
+  let s = Whenstate.copy_prev_state !state s in
   state := s;
 
   (* Re-evaluate all when jobs. *)
-  reevaluate_whenjobs ~onload:true (Whenstate.get_whenjobs !state);
+  let jobs = Whenstate.get_whenjobs !state in
+  let jobnames, state' = reevaluate_whenjobs ~onload:true !state jobs in
+  let state' = run_whenjobs state' jobnames in
+  state := state';
 
   (* Schedule the next every job to run. *)
   schedule_next_everyjob ()
 
 (* Re-evaluate each when-statement job, in a loop until we reach
- * a fixpoint.  Run those that need to be run.
+ * a fixpoint.  Return the list of job names that should run and
+ * the updated state.
  *)
-and reevaluate_whenjobs ?onload jobs =
-  let rec loop set jobs =
-    let set' =
+and reevaluate_whenjobs ?onload state jobs =
+  let rec loop (set, state) jobs =
+    let set', state' =
       List.fold_left (
-        fun set job ->
+        fun (set, state) job ->
           let r, state' =
-            try Whenstate.evaluate_whenjob ?onload !state job
+            try Whenstate.evaluate_whenjob ?onload state job
             with Invalid_argument err | Failure err ->
               Syslog.error "error evaluating job %s (at %s): %s"
                 job.job_name (Camlp4.PreCast.Ast.Loc.to_string job.job_loc) err;
-              false, !state in
-
-          state := state';
+              false, state in
 
           if !debug then
             Syslog.notice "evaluate %s -> %b\n" job.job_name r;
 
-          if r then StringSet.add job.job_name set else set
-      ) set jobs in
+          (if r then StringSet.add job.job_name set else set), state'
+      ) (set, state) jobs in
+    (* reached a fixpoint? *)
     if StringSet.compare set set' <> 0 then
-      loop set' jobs
+      loop (set', state') jobs
     else
-      set'
+      (set', state')
   in
-  let set = loop StringSet.empty jobs in
+  let set, state = loop (StringSet.empty, state) jobs in
   let jobnames = StringSet.elements set in
 
   (* Ensure the jobs always run in predictable (name) order. *)
   let jobnames = List.sort compare_jobnames jobnames in
+  jobnames, state
 
+and run_whenjobs state jobnames =
   (* Run the jobs. *)
-  List.iter run_job (List.map (Whenstate.get_job !state) jobnames)
+  let jobs = List.map (Whenstate.get_job state) jobnames in
+  List.fold_left run_job state jobs
 
 (* Schedule the next every-statement job to run, if there is one.  We
  * look at the every jobs, work out the time that each must run at,
@@ -299,7 +439,8 @@ and schedule_next_everyjob () =
       let t_diff = if t_diff < 0. then 0. else t_diff in
       let run_jobs () =
         delete_timer_group ();          (* Delete the timer. *)
-        List.iter run_job jobs;
+        let state' = List.fold_left run_job !state jobs in
+        state := state';
         schedule_next_everyjob ()
       in
       Unixqueue.weak_once esys g t_diff run_jobs;
@@ -319,67 +460,101 @@ and delete_timer_group () =
     Unixqueue.clear esys g;
     timer_group := None
 
-and run_job job =
+and run_job state job =
   (* Increment JOBSERIAL. *)
-  let serial =
-    match Whenstate.get_variable !state "JOBSERIAL" with
+  let serial, state =
+    match Whenstate.get_variable state "JOBSERIAL" with
     | T_int serial ->
       let serial = succ_big_int serial in
-      state := Whenstate.set_variable !state "JOBSERIAL" (T_int serial);
-      serial
+      let state' = Whenstate.set_variable state "JOBSERIAL" (T_int serial) in
+      serial, state'
     | _ -> assert false in
 
-  Syslog.notice "running %s (JOBSERIAL=%s)"
-    job.job_name (string_of_big_int serial);
-
-  (* Create a temporary directory.  The current directory of the job
-   * will be in this directory.  The directory is removed when the
-   * child process exits.
+  (* Call the pre-condition script.  Note this may decide not to run
+   * the job by returning false.
    *)
-  let dir = tmpdir () in
-
-  let pid = fork () in
-  if pid = 0 then ( (* child process running the job *)
-    chdir dir;
-
-    (* Set environment variables corresponding to each variable. *)
-    List.iter
-      (fun (name, value) -> putenv name (string_of_variable value))
-      (Whenstate.get_variables !state);
-
-    (* Set the $JOBNAME environment variable. *)
-    putenv "JOBNAME" job.job_name;
-
-    (* Create a temporary file containing the shell script fragment. *)
-    let script = dir // "script.sh" in
-    let chan = open_out script in
-    fprintf chan "set -e\n"; (* So that jobs exit on error. *)
-    output_string chan job.job_script.sh_script;
-    close_out chan;
-    chmod script 0o700;
-
-    let shell = try getenv "SHELL" with Not_found -> "/bin/sh" in
-
-    (* Set output to file. *)
-    let output = dir // "output.txt" in
-    let fd = openfile output [O_WRONLY; O_CREAT; O_TRUNC; O_NOCTTY] 0o600 in
-    dup2 fd stdout;
-    dup2 fd stderr;
-    close fd;
-
-    (* Execute the shell script. *)
-    (try execvp shell [| shell; "-c"; script |];
-     with Unix_error (err, fn, _) ->
-       Syslog.error "%s failed: %s: %s" fn script (error_message err)
+  let pre_condition () =
+    match job.job_pre with
+    | None -> true
+    | Some pre ->
+      let rs = ref [] in
+      IntMap.iter (
+        fun pid (job, _, serial, start_time) ->
+          let r = { pirun_job_name = job.job_name;
+                    pirun_serial = serial;
+                    pirun_start_time = start_time;
+                    pirun_pid = pid } in
+          rs := r :: !rs
+      ) !runningmap;
+      let preinfo = {
+        pi_job_name = job.job_name;
+        pi_serial = serial;
+        pi_variables = Whenstate.get_variables state;
+        pi_running = !rs;
+      } in
+      pre preinfo
+  in
+  if pre_condition () then (
+    Syslog.notice "running %s (JOBSERIAL=%s)"
+      job.job_name (string_of_big_int serial);
+
+    (* Create a temporary directory.  The current directory of the job
+     * will be in this directory.  The directory is removed when the
+     * child process exits.
+     *)
+    let dir = tmpdir () in
+
+    let pid = fork () in
+    if pid = 0 then ( (* child process running the job *)
+      chdir dir;
+
+      (* Set environment variables corresponding to each variable. *)
+      List.iter
+        (fun (name, value) -> putenv name (string_of_variable value))
+        (Whenstate.get_variables state);
+
+      (* Set the $JOBNAME environment variable. *)
+      putenv "JOBNAME" job.job_name;
+
+      (* Create a temporary file containing the shell script fragment. *)
+      let script = dir // "script.sh" in
+      let chan = open_out script in
+      fprintf chan "set -e\n"; (* So that jobs exit on error. *)
+      output_string chan job.job_script.sh_script;
+      close_out chan;
+      chmod script 0o700;
+
+      let shell = try getenv "SHELL" with Not_found -> "/bin/sh" in
+
+      (* Set output to file. *)
+      let output = dir // "output.txt" in
+      let fd = openfile output [O_WRONLY; O_CREAT; O_TRUNC; O_NOCTTY] 0o600 in
+      dup2 fd stdout;
+      dup2 fd stderr;
+      close fd;
+
+      (* Execute the shell script. *)
+      (try execvp shell [| shell; "-c"; script |];
+       with Unix_error (err, fn, _) ->
+         Syslog.error "%s failed: %s: %s" fn script (error_message err)
+      );
+      _exit 1
     );
-    _exit 1
-  );
 
-  (* Remember this PID, the job and the temporary directory, so we
-   * can clean up when the child exits.
-   *)
-  runningmap := IntMap.add pid (job, dir, serial, time ()) !runningmap;
-  serialmap := BigIntMap.add serial pid !serialmap
+    (* Remember this PID, the job and the temporary directory, so we
+     * can clean up when the child exits.
+     *)
+    runningmap := IntMap.add pid (job, dir, serial, time ()) !runningmap;
+    serialmap := BigIntMap.add serial pid !serialmap;
+
+    state
+  )
+  else (
+    Syslog.notice "not running %s (JOBSERIAL=%s) because pre() condition returned false"
+      job.job_name (string_of_big_int serial);
+
+    state
+  )
 
 and tmpdir () =
   let chan = open_in "/dev/urandom" in