X-Git-Url: http://git.annexia.org/?p=whenjobs.git;a=blobdiff_plain;f=daemon%2Fdaemon.ml;h=542c7a48b40ffb255708c766928ca430616aab82;hp=a6cc2fa7e75346756efdcc295fb17c12652af71f;hb=083f42734bf06c6a752e3a93e519c6250a04dd96;hpb=28d4576308b10064eda39827c419aa33e1041041 diff --git a/daemon/daemon.ml b/daemon/daemon.ml index a6cc2fa..542c7a4 100644 --- a/daemon/daemon.ml +++ b/daemon/daemon.ml @@ -17,31 +17,21 @@ *) open Whenutils +open Whenexpr +open Big_int open Unix open Printf (* See [exit.c]. *) external _exit : int -> 'a = "whenjobs__exit" -(* All jobs that are loaded. Maps name -> [job] structure. *) -let jobs = ref StringMap.empty - -(* Map variable names to jobs which depend on that variable. This - * gives us a quick way to tell which jobs might need to be reevaluated - * when a variable is set. - *) -let dependencies = ref StringMap.empty - -(* Current values of variables. Using the referentially transparent - * type Map is very useful here because it lets us cheaply keep - * previous values of variables. - *) -let variables : variables ref = ref StringMap.empty - (* $HOME/.whenjobs *) let jobsdir = ref "" +(* The state. *) +let state = ref Whenstate.empty + (* Jobs that are running; map of PID -> (job, other data). Note that * the job may no longer exist *OR* it may have been renamed, * eg. if the jobs file was reloaded. @@ -56,6 +46,14 @@ let server = ref None let esys = Unixqueue.standard_event_system () +(* The timer. It's convenient to have this as a global variable + * because (a) there should only be one timer (which fires when the + * soonest every-job becomes ready), and (b) it's complicated to track + * that timer and avoid it getting double-scheduled (eg. when we + * reload the jobs file) without having a global variable. + *) +let timer_group = ref None + let rec init j d = jobsdir := j; debug := d; @@ -66,6 +64,7 @@ let rec init j d = let addr = sprintf "%s/socket" !jobsdir in (try unlink addr with Unix_error _ -> ()); + (* Create the Unix domain socket server. *) server := Some ( Whenproto_srv.When.V1.create_server ~proc_reload_file @@ -80,7 +79,10 @@ let rec init j d = ); (* Handle SIGCHLD to clean up jobs. *) - Sys.set_signal Sys.sigchld (Sys.Signal_handle handle_sigchld) + Sys.set_signal Sys.sigchld (Sys.Signal_handle handle_sigchld); + + (* Initialize the variables. *) + state := Whenstate.set_variable !state "JOBSERIAL" (T_int zero_big_int) and proc_reload_file () = if !debug then Syslog.notice "remote call: reload_file"; @@ -91,27 +93,31 @@ and proc_reload_file () = and proc_set_variable (name, value) = if !debug then Syslog.notice "remote call: set_variable %s" name; - let value = variable_of_rpc value in - variables := StringMap.add name value !variables; + try + check_valid_variable_name name; + + let value = variable_of_rpc value in + 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; - (* Which jobs need to be re-evaluated? *) - let jobnames = try StringMap.find name !dependencies with Not_found -> [] in - reevaluate_whenjobs jobnames + `ok + with + Failure msg -> `error msg and proc_get_variable name = if !debug then Syslog.notice "remote call: get_variable %s" name; - try rpc_of_variable (StringMap.find name !variables) - with (* all non-existent variables are empty strings *) - Not_found -> `string_t "" + rpc_of_variable (Whenstate.get_variable !state name) and proc_get_variable_names () = if !debug then Syslog.notice "remote call: get_variable_names"; - (* Only return variables that are non-empty. *) - let vars = StringMap.fold ( - fun name value xs -> if value <> T_string "" then name :: xs else xs - ) !variables [] in + let vars = Whenstate.get_variable_names !state in + + (* Return variable names as a sorted array. *) let vars = Array.of_list vars in Array.sort compare vars; vars @@ -130,14 +136,20 @@ and proc_exit_daemon () = (* Reload the jobs file. *) and reload_file () = let file = sprintf "%s/jobs.cmo" !jobsdir in - Whenfile.init (); - let js = + (* 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 + * state. + *) + let s = Whenstate.copy_variables !state Whenstate.empty in + Whenfile.init s; + + let s = try Dynlink.loadfile file; - let jobs = Whenfile.get_jobs () in - Syslog.notice "loaded %d job(s) from %s" (List.length jobs) file; - jobs + let s = Whenfile.get_state () in + Syslog.notice "loaded %d job(s) from %s" (Whenstate.nr_jobs s) file; + s with | Dynlink.Error err -> let err = Dynlink.error_message err in @@ -146,73 +158,49 @@ and reload_file () = | exn -> failwith (Printexc.to_string exn) in - (* Set 'jobs' and related global variables. *) - let () = - let map = List.fold_left ( - fun map j -> - let name = j.job_name in - StringMap.add name j map - ) StringMap.empty js in - jobs := map in - - let () = - let map = List.fold_left ( - fun map j -> - let deps = dependencies_of_job j in - let name = j.job_name in - List.fold_left ( - fun map d -> - let names = try StringMap.find d map with Not_found -> [] in - StringMap.add d (name :: names) map - ) map deps - ) StringMap.empty js in - dependencies := map in + state := s; (* Re-evaluate all when jobs. *) - reevaluate_whenjobs (StringMap.keys !jobs); + reevaluate_whenjobs ~onload:true (Whenstate.get_whenjobs !state); (* Schedule the next every job to run. *) schedule_next_everyjob () -(* Re-evaluate each named when-statement job, in a loop until we reach - * a fixpoint. Run those that need to be run. every-statement jobs - * are ignored here. +(* Re-evaluate each when-statement job, in a loop until we reach + * a fixpoint. Run those that need to be run. *) -and reevaluate_whenjobs jobnames = - let rec loop set jobnames = +and reevaluate_whenjobs ?onload jobs = + let rec loop set jobs = let set' = List.fold_left ( - fun set jobname -> - let job = - try StringMap.find jobname !jobs - with Not_found -> assert false in - assert (jobname = job.job_name); - - let r, job' = - try job_evaluate job !variables + fun set job -> + let r, state' = + try Whenstate.evaluate_whenjob ?onload !state job with Invalid_argument err | Failure err -> Syslog.error "error evaluating job %s (at %s): %s" - jobname (Camlp4.PreCast.Ast.Loc.to_string job.job_loc) err; - false, job in + job.job_name (Camlp4.PreCast.Ast.Loc.to_string job.job_loc) err; + false, !state in - jobs := StringMap.add jobname job' !jobs; + state := state'; if !debug then - Syslog.notice "evaluate %s -> %b\n" jobname r; + Syslog.notice "evaluate %s -> %b\n" job.job_name r; - if r then StringSet.add jobname set else set - ) set jobnames in + if r then StringSet.add job.job_name set else set + ) set jobs in if StringSet.compare set set' <> 0 then - loop set' jobnames + loop set' jobs else set' in - let set = loop StringSet.empty jobnames in + let set = loop StringSet.empty 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 - List.iter run_job - (List.map (fun jobname -> StringMap.find jobname !jobs) jobnames) + + (* Run the jobs. *) + List.iter run_job (List.map (Whenstate.get_job !state) jobnames) (* 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, @@ -224,11 +212,11 @@ and schedule_next_everyjob () = let t = time () in (* Get only everyjobs. *) - let jobs = StringMap.values !jobs in - let jobs = filter_map ( + let jobs = Whenstate.get_everyjobs !state in + let jobs = List.map ( function - | { job_cond = Every_job period } as job -> Some (job, period) - | { job_cond = When_job _ } -> None + | { job_cond = Every_job period } as job -> (job, period) + | { job_cond = When_job _ } -> assert false ) jobs in (* Map everyjob to next time it must run. *) @@ -272,11 +260,11 @@ and schedule_next_everyjob () = (string_of_time_t t); (* Schedule them to run at time t. *) - let g = Unixqueue.new_group esys in + let g = new_timer_group () in let t_diff = t -. Unix.time () in let t_diff = if t_diff < 0. then 0. else t_diff in let run_jobs () = - Unixqueue.clear esys g; (* Delete the timer. *) + delete_timer_group (); (* Delete the timer. *) List.iter run_job jobs; schedule_next_everyjob () in @@ -284,6 +272,19 @@ and schedule_next_everyjob () = ) ) +and new_timer_group () = + delete_timer_group (); + let g = Unixqueue.new_group esys in + timer_group := Some g; + g + +and delete_timer_group () = + match !timer_group with + | None -> () + | Some g -> + Unixqueue.clear esys g; + timer_group := None + and string_of_time_t t = let tm = gmtime t in sprintf "%04d-%02d-%02d %02d:%02d:%02d UTC" @@ -291,7 +292,18 @@ and string_of_time_t t = tm.tm_hour tm.tm_min tm.tm_sec and run_job job = - Syslog.notice "running %s" job.job_name; + let () = + (* Increment JOBSERIAL. *) + let serial = + 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 + | _ -> assert false in + + Syslog.notice "running %s (JOBSERIAL=%s)" + job.job_name (string_of_big_int serial) in (* Create a temporary directory. The current directory of the job * will be in this directory. The directory is removed when the @@ -304,8 +316,9 @@ and run_job job = chdir dir; (* Set environment variables corresponding to each variable. *) - StringMap.iter - (fun name value -> putenv name (string_of_variable value)) !variables; + 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; @@ -313,12 +326,15 @@ and run_job job = (* Create a temporary file containing the shell script fragment. *) let script = dir // "script" 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 + (* Execute the shell script. *) - (try execvp "bash" [| "bash"; "-c"; script |]; + (try execvp shell [| shell; "-c"; script |]; with Unix_error (err, fn, _) -> Syslog.error "%s failed: %s: %s" fn script (error_message err) );