2 * Copyright (C) 2012 Red Hat Inc.
4 * This program is free software; you can redistribute it and/or modify
5 * it under the terms of the GNU General Public License as published by
6 * the Free Software Foundation; either version 2 of the License, or
7 * (at your option) any later version.
9 * This program is distributed in the hope that it will be useful,
10 * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 * GNU General Public License for more details.
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
24 (* All jobs that are loaded. Maps name -> [job] structure. *)
25 let jobs = ref StringMap.empty
27 (* Map variable names to jobs which depend on that variable. This
28 * gives us a quick way to tell which jobs might need to be reevaluated
29 * when a variable is set.
31 let dependencies = ref StringMap.empty
33 (* Current values of variables. Using the referentially transparent
34 * type Map is very useful here because it lets us cheaply keep
35 * previous values of variables.
37 let variables : variables ref = ref StringMap.empty
39 (* Last time that an every job ran. See schedule_next_everyjob below. *)
40 let last_t = ref (time ())
45 (* Was debugging requested on the command line? *)
51 let esys = Unixqueue.standard_event_system ()
52 let timer_group = ref None
58 Whenlock.create_lock !jobsdir;
60 (* Remove old socket if it exists. *)
61 let addr = sprintf "%s/socket" !jobsdir in
62 (try unlink addr with Unix_error _ -> ());
65 Whenproto_srv.When.V1.create_server
69 ~proc_get_variable_names
71 (Rpc_server.Unix addr)
72 Rpc.Tcp (* not TCP, this is the same as SOCK_STREAM *)
77 and proc_reload_file () =
78 if !debug then Syslog.notice "remote call: reload_file";
80 try reload_file (); `ok
81 with Failure err -> `error err
83 and proc_set_variable (name, value) =
84 if !debug then Syslog.notice "remote call: set_variable %s" name;
86 let value = variable_of_rpc value in
87 variables := StringMap.add name value !variables;
89 (* Which jobs need to be re-evaluated? *)
90 let jobnames = try StringMap.find name !dependencies with Not_found -> [] in
91 reevaluate_whenjobs jobnames
93 and proc_get_variable name =
94 if !debug then Syslog.notice "remote call: get_variable %s" name;
96 try rpc_of_variable (StringMap.find name !variables)
97 with (* all non-existent variables are empty strings *)
98 Not_found -> `string_t ""
100 and proc_get_variable_names () =
101 if !debug then Syslog.notice "remote call: get_variable_names";
103 (* Only return variables that are non-empty. *)
104 let vars = StringMap.fold (
105 fun name value xs -> if value <> T_string "" then name :: xs else xs
107 let vars = Array.of_list vars in
108 Array.sort compare vars;
111 and proc_exit_daemon () =
112 if !debug then Syslog.notice "remote call: exit_daemon";
116 `error "exit_daemon: no server handle"
118 Rpc_server.stop_server ~graceful:true s;
122 (* Reload the jobs file. *)
124 let file = sprintf "%s/jobs.cmo" !jobsdir in
129 Dynlink.loadfile file;
130 let jobs = Whenfile.get_jobs () in
131 Syslog.notice "loaded %d job(s) from %s" (List.length jobs) file;
134 | Dynlink.Error err ->
135 let err = Dynlink.error_message err in
136 Syslog.error "error loading jobs: %s" err;
139 failwith (Printexc.to_string exn) in
141 (* Set 'jobs' and related global variables. *)
143 let map = List.fold_left (
145 let name = j.job_name in
146 StringMap.add name j map
147 ) StringMap.empty js in
151 let map = List.fold_left (
153 let deps = dependencies_of_job j in
154 let name = j.job_name in
157 let names = try StringMap.find d map with Not_found -> [] in
158 StringMap.add d (name :: names) map
160 ) StringMap.empty js in
161 dependencies := map in
163 (* Re-evaluate all when jobs. *)
164 reevaluate_whenjobs (StringMap.keys !jobs);
166 (* Schedule the next every job to run. *)
168 schedule_next_everyjob ()
170 (* Re-evaluate each named when-statement job, in a loop until we reach
171 * a fixpoint. Run those that need to be run. every-statement jobs
174 and reevaluate_whenjobs jobnames =
175 let rec loop set jobnames =
180 try StringMap.find jobname !jobs
181 with Not_found -> assert false in
182 assert (jobname = job.job_name);
185 try job_evaluate job !variables
186 with Invalid_argument err | Failure err ->
187 Syslog.error "error evaluating job %s (at %s): %s"
188 jobname (Camlp4.PreCast.Ast.Loc.to_string job.job_loc) err;
191 jobs := StringMap.add jobname job' !jobs;
194 Syslog.notice "evaluate %s -> %b\n" jobname r;
196 if r then StringSet.add jobname set else set
198 if StringSet.compare set set' <> 0 then
203 let set = loop StringSet.empty jobnames in
204 let jobnames = StringSet.elements set in
205 (* Ensure the jobs always run in predictable (name) order. *)
206 let jobnames = List.sort compare jobnames in
208 (List.map (fun jobname -> StringMap.find jobname !jobs) jobnames)
210 (* Schedule the next every-statement job to run, if there is one. We
211 * look at the every jobs, work out the time that each must run at,
212 * pick the job(s) which must run soonest, and schedule a timer to run
213 * them. When the timer fires, it runs those jobs, then call this
214 * function again. 'last_t' is the base time used for scheduling (or
215 * the time that the file was last reloaded).
217 and schedule_next_everyjob () =
218 (* Get only everyjobs. *)
219 let jobs = StringMap.values !jobs in
220 let jobs = filter_map (
222 | { job_cond = Every_job period } as job -> Some (job, period)
223 | { job_cond = When_job _ } -> None
226 (* Map everyjob to next time it must run. *)
227 let jobs = List.map (
229 let t' = next_periodexpr !last_t period in
230 assert (t' > !last_t); (* serious bug in next_periodexpr if false *)
234 (* Sort, soonest first. *)
235 let jobs = List.sort (fun (_,a) (_,b) -> compare a b) jobs in
240 Syslog.notice "%s: next scheduled run at %s"
241 job.job_name (string_of_time_t t)
245 (* Pick the job(s) which run soonest. *)
246 let rec pick = function
249 | (j1, t) :: (j2, t') :: _ when t < t' -> t, [j1]
250 | (j1, t) :: (((j2, t') :: _) as rest) -> t, (j1 :: snd (pick rest))
252 let t, jobs = pick jobs in
259 Syslog.notice "scheduling job(s) %s to run at %s"
260 (String.concat ", " (List.map (fun { job_name = name } -> name) jobs))
261 (string_of_time_t t);
263 (* Schedule them to run at time t. *)
264 let g = new_timer_group () in
265 let w = Unixqueue.new_wait_id esys in
266 let t_diff = t -. Unix.time () in
267 let t_diff = if t_diff < 0. then 0. else t_diff in
268 Unixqueue.add_weak_resource esys g (Unixqueue.Wait w, t_diff);
270 List.iter run_job jobs;
272 schedule_next_everyjob ();
274 Unixqueue.add_handler esys g run_jobs;
278 and string_of_time_t t =
280 sprintf "%04d-%02d-%02d %02d:%02d:%02d UTC"
281 (1900+tm.tm_year) (1+tm.tm_mon) tm.tm_mday
282 tm.tm_hour tm.tm_min tm.tm_sec
284 and new_timer_group () =
286 let g = Unixqueue.new_group esys in
287 timer_group := Some g;
290 and delete_timer () =
291 match !timer_group with
294 Unixqueue.clear esys g;
298 Syslog.notice "running %s" job.job_name;