X-Git-Url: http://git.annexia.org/?p=whenjobs.git;a=blobdiff_plain;f=daemon%2Fdaemon.ml;h=bc4f51a2f1177e5be2692a233d7f2b850f9a55bc;hp=d9393e74982c9549b042d635db701dbf5e4414dc;hb=ea28f8eebc4d8434e7e66d62b769d747656d27ae;hpb=6c41dfae8a0d60cab2fd3b82f73b762972496c2c

diff --git a/daemon/daemon.ml b/daemon/daemon.ml
index d9393e7..bc4f51a 100644
--- a/daemon/daemon.ml
+++ b/daemon/daemon.ml
@@ -81,6 +81,9 @@ let rec init j d =
       ~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
@@ -96,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) =
@@ -110,7 +113,8 @@ and proc_set_variable (name, value) =
 
     (* Which jobs need to be re-evaluated? *)
     let jobs = Whenstate.get_dependencies !state [name] in
-    let state' = reevaluate_whenjobs !state jobs in
+    let jobnames, state' = reevaluate_whenjobs !state jobs in
+    let state' = run_whenjobs state' jobnames in
     state := state';
 
     `ok
@@ -210,7 +214,8 @@ and proc_set_variables vars =
 
     (* Which jobs need to be re-evaluated? *)
     let jobs = Whenstate.get_dependencies !state (List.map fst vars) in
-    let state' = reevaluate_whenjobs !state jobs in
+    let jobnames, state' = reevaluate_whenjobs !state jobs in
+    let state' = run_whenjobs state' jobnames in
     state := state';
 
     `ok
@@ -220,9 +225,80 @@ and proc_set_variables vars =
 and proc_get_job_names () =
   Array.of_list (Whenstate.get_job_names !state)
 
-(* Reload the jobs file. *)
-and reload_file () =
-  let file = sprintf "%s/jobs.cmo" !jobsdir in
+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 (bytecode) or dir/*.cmxs (native code) *)
+  let suffix = if not Dynlink.is_native then ".cmo" else ".cmxs" in
+  let dir = !jobsdir in
+  let files = Array.to_list (Sys.readdir dir) in
+  let files = List.filter (fun file -> string_endswith file suffix) 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
@@ -233,9 +309,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 ->
@@ -250,14 +327,16 @@ and reload_file () =
 
   (* Re-evaluate all when jobs. *)
   let jobs = Whenstate.get_whenjobs !state in
-  let state' = reevaluate_whenjobs ~onload:true !state jobs 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 the jobs and return the updated state.
+ * a fixpoint.  Return the list of job names that should run and
+ * the updated state.
  *)
 and reevaluate_whenjobs ?onload state jobs =
   let rec loop (set, state) jobs =
@@ -287,7 +366,9 @@ and reevaluate_whenjobs ?onload state jobs =
 
   (* 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. *)
   let jobs = List.map (Whenstate.get_job state) jobnames in
   List.fold_left run_job state jobs