80f2cbd487f4cf35ad1a4b521305742db1383d92
[whenjobs.git] / daemon / daemon.ml
1 (* whenjobs
2  * Copyright (C) 2012 Red Hat Inc.
3  *
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.
8  *
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.
13  *
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.
17  *)
18
19 open Whenutils
20
21 open Unix
22 open Printf
23
24 (* All jobs that are loaded.  Maps name -> [job] structure. *)
25 let jobs = ref StringMap.empty
26
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.
30  *)
31 let dependencies = ref StringMap.empty
32
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.
36  *)
37 let variables : variables ref = ref StringMap.empty
38
39 (* Last time that an every job ran.  See schedule_next_everyjob below. *)
40 let last_t = ref (time ())
41
42 (* $HOME/.whenjobs *)
43 let jobsdir = ref ""
44
45 (* Was debugging requested on the command line? *)
46 let debug = ref false
47
48 let esys = Unixqueue.standard_event_system ()
49 let timer_group = ref None
50
51 let rec init j d =
52   jobsdir := j;
53   debug := d;
54
55   Whenlock.create_lock !jobsdir;
56
57   (* Remove old socket if it exists. *)
58   let addr = sprintf "%s/socket" !jobsdir in
59   (try unlink addr with Unix_error _ -> ());
60
61   ignore (
62     Whenproto_srv.When.V1.create_server
63       ~proc_reload_file
64       ~proc_set_variable
65       ~proc_get_variable
66       ~proc_get_variable_names
67       (Rpc_server.Unix addr)
68       Rpc.Tcp (* not TCP, this is the same as SOCK_STREAM *)
69       Rpc.Socket
70       esys
71   )
72
73 and proc_reload_file () =
74   if !debug then Syslog.notice "remote call: reload_file";
75
76   try reload_file (); `ok
77   with Failure err -> `error err
78
79 and proc_set_variable (name, value) =
80   if !debug then Syslog.notice "remote call: set_variable %s" name;
81
82   let value = variable_of_rpc value in
83   variables := StringMap.add name value !variables;
84
85   (* Which jobs need to be re-evaluated? *)
86   let jobnames = try StringMap.find name !dependencies with Not_found -> [] in
87   reevaluate_whenjobs jobnames
88
89 and proc_get_variable name =
90   if !debug then Syslog.notice "remote call: get_variable %s" name;
91
92   try rpc_of_variable (StringMap.find name !variables)
93   with (* all non-existent variables are empty strings *)
94     Not_found -> `string_t ""
95
96 and proc_get_variable_names () =
97   if !debug then Syslog.notice "remote call: get_variable_names";
98
99   (* Only return variables that are non-empty. *)
100   let vars = StringMap.fold (
101     fun name value xs -> if value <> T_string "" then name :: xs else xs
102   ) !variables [] in
103   let vars = Array.of_list vars in
104   Array.sort compare vars;
105   vars
106
107 (* Reload the jobs file. *)
108 and reload_file () =
109   let file = sprintf "%s/jobs.cmo" !jobsdir in
110   Whenfile.init ();
111
112   let js =
113     try
114       Dynlink.loadfile file;
115       let jobs = Whenfile.get_jobs () in
116       Syslog.notice "loaded %d job(s) from %s" (List.length jobs) file;
117       jobs
118     with
119     | Dynlink.Error err ->
120       let err = Dynlink.error_message err in
121       Syslog.error "error loading jobs: %s" err;
122       failwith err
123     | exn ->
124       failwith (Printexc.to_string exn) in
125
126   (* Set 'jobs' and related global variables. *)
127   let () =
128     let map = List.fold_left (
129       fun map j ->
130         let name = j.job_name in
131         StringMap.add name j map
132     ) StringMap.empty js in
133     jobs := map in
134
135   let () =
136     let map = List.fold_left (
137       fun map j ->
138         let deps = dependencies_of_job j in
139         let name = j.job_name in
140         List.fold_left (
141           fun map d ->
142             let names = try StringMap.find d map with Not_found -> [] in
143             StringMap.add d (name :: names) map
144         ) map deps
145     ) StringMap.empty js in
146     dependencies := map in
147
148   (* Re-evaluate all when jobs. *)
149   reevaluate_whenjobs (StringMap.keys !jobs);
150
151   (* Schedule the next every job to run. *)
152   last_t := time ();
153   schedule_next_everyjob ()
154
155 (* Re-evaluate each named when-statement job, in a loop until we reach
156  * a fixpoint.  Run those that need to be run.  every-statement jobs
157  * are ignored here.
158  *)
159 and reevaluate_whenjobs jobnames =
160   let rec loop set jobnames =
161     let set' =
162       List.fold_left (
163         fun set jobname ->
164           let job =
165             try StringMap.find jobname !jobs
166             with Not_found -> assert false in
167           assert (jobname = job.job_name);
168
169           let r, job' = job_evaluate job !variables in
170           jobs := StringMap.add jobname job' !jobs;
171
172           if !debug then
173             Syslog.notice "evaluate %s -> %b\n" jobname r;
174
175           if r then StringSet.add jobname set else set
176       ) set jobnames in
177     if StringSet.compare set set' <> 0 then
178       loop set' jobnames
179     else
180       set'
181   in
182   let set = loop StringSet.empty jobnames in
183   let jobnames = StringSet.elements set in
184   List.iter run_job
185     (List.map (fun jobname -> StringMap.find jobname !jobs) jobnames)
186
187 (* Schedule the next every-statement job to run, if there is one.  We
188  * look at the every jobs, work out the time that each must run at,
189  * pick the job(s) which must run soonest, and schedule a timer to run
190  * them.  When the timer fires, it runs those jobs, then call this
191  * function again.  'last_t' is the base time used for scheduling (or
192  * the time that the file was last reloaded).
193  *)
194 and schedule_next_everyjob () =
195   (* Get only everyjobs. *)
196   let jobs = StringMap.values !jobs in
197   let jobs = filter_map (
198     function
199     | { job_cond = Every_job period } as job -> Some (job, period)
200     | { job_cond = When_job _ } -> None
201   ) jobs in
202
203   (* Map everyjob to next time it must run. *)
204   let jobs = List.map (
205     fun (job, period) ->
206       let t' = next_periodexpr !last_t period in
207       assert (t' > !last_t); (* serious bug in next_periodexpr if false *)
208       job, t'
209   ) jobs in
210
211   (* Sort, soonest first. *)
212   let jobs = List.sort (fun (_,a) (_,b) -> compare a b) jobs in
213
214   if !debug then (
215     List.iter (
216       fun (job, t) ->
217         Syslog.notice "%s: next scheduled run at %s"
218           job.job_name (string_of_time_t t)
219     ) jobs
220   );
221
222   (* Pick the job(s) which run soonest. *)
223   let rec pick = function
224     | [] -> 0., []
225     | [j, t] -> t, [j]
226     | (j1, t) :: (j2, t') :: _ when t < t' -> t, [j1]
227     | (j1, t) :: (((j2, t') :: _) as rest) -> t, (j1 :: snd (pick rest))
228   in
229   let t, jobs = pick jobs in
230
231   if t > 0. then (
232     last_t := t;
233
234     if jobs <> [] then (
235       if !debug then
236         Syslog.notice "scheduling job(s) %s to run at %s"
237           (String.concat ", " (List.map (fun { job_name = name } -> name) jobs))
238           (string_of_time_t t);
239
240       (* Schedule them to run at time t. *)
241       let g = new_timer_group () in
242       let w = Unixqueue.new_wait_id esys in
243       let t_diff = t -. Unix.time () in
244       let t_diff = if t_diff < 0. then 0. else t_diff in
245       Unixqueue.add_resource esys g (Unixqueue.Wait w, t_diff);
246       let run_jobs _ _ _ =
247         List.iter run_job jobs;
248         delete_timer ();
249         schedule_next_everyjob ();
250       in
251       Unixqueue.add_handler esys g run_jobs;
252     )
253   )
254
255 and string_of_time_t t =
256   let tm = gmtime t in
257   sprintf "%04d-%02d-%02d %02d:%02d:%02d UTC"
258     (1900+tm.tm_year) (1+tm.tm_mon) tm.tm_mday
259     tm.tm_hour tm.tm_min tm.tm_sec
260
261 and new_timer_group () =
262   delete_timer ();
263   let g = Unixqueue.new_group esys in
264   timer_group := Some g;
265   g
266
267 and delete_timer () =
268   match !timer_group with
269   | None -> ()
270   | Some g ->
271     Unixqueue.clear esys g;
272     timer_group := None
273
274 and run_job job =
275   Syslog.notice "running %s" job.job_name;
276   () (* XXX *)
277
278 let main_loop () =
279   Unixqueue.run esys