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 open Whenexpr
  21
  22 open Big_int
  23 open Unix
  24 open Printf
  25
  26 (* See [exit.c]. *)
  27 external _exit : int -> 'a = "whenjobs__exit"
  28
  29 (* $HOME/.whenjobs *)
  30 let jobsdir = ref ""
  31
  32 (* The state.
  33  *
  34  * Note that whenever this is updated, you need to consider if you
  35  * should call 'save_variables ()' (which persists the variables to a
  36  * file).  XXX We should replace this ref with an accessor
  37  * function.
  38  *)
  39 let state = ref Whenstate.empty
  40
  41 (* Format used to save variables.  Note we can't allow any internal
  42  * types to "escape" into this definition, else the file format will
  43  * change when parts of the program change.
  44  *)
  45 type variables_file_v1 = (string * variable_v1) list
  46 and variable_v1 =
  47   | Vv1_unit
  48   | Vv1_bool of bool
  49   | Vv1_string of string
  50   | Vv1_int of big_int
  51   | Vv1_float of float
  52
  53 let variable_of_variable_v1 = function
  54   | Vv1_unit -> T_unit
  55   | Vv1_bool b -> T_bool b
  56   | Vv1_string s -> T_string s
  57   | Vv1_int i -> T_int i
  58   | Vv1_float f -> T_float f
  59
  60 let variable_v1_of_variable = function
  61   | T_unit -> Vv1_unit
  62   | T_bool b -> Vv1_bool b
  63   | T_string s -> Vv1_string s
  64   | T_int i -> Vv1_int i
  65   | T_float f -> Vv1_float f
  66
  67 (* Jobs that are running: a map of PID -> (job, tmpdir, serial, start_time).
  68  * Note that the job may no longer exist *OR* it may have been renamed,
  69  * eg. if the jobs file was reloaded.
  70  *)
  71 let runningmap = ref IntMap.empty
  72
  73 (* Serial numbers of running jobs.  Map of serial -> PID (in runningmap). *)
  74 let serialmap = ref BigIntMap.empty
  75
  76 (* Was debugging requested on the command line? *)
  77 let debug = ref false
  78
  79 (* The server. *)
  80 let server = ref None
  81
  82 let esys = Unixqueue.standard_event_system ()
  83
  84 (* The timer.  It's convenient to have this as a global variable
  85  * because (a) there should only be one timer (which fires when the
  86  * soonest every-job becomes ready), and (b) it's complicated to track
  87  * that timer and avoid it getting double-scheduled (eg.  when we
  88  * reload the jobs file) without having a global variable.
  89  *)
  90 let timer_group = ref None
  91
  92 let rec init j d =
  93   jobsdir := j;
  94   debug := d;
  95
  96   Whenlock.create_lock !jobsdir;
  97
  98   (* Remove old socket if it exists. *)
  99   let addr = sprintf "%s/socket" !jobsdir in
 100   (try unlink addr with Unix_error _ -> ());
 101
 102   (* Create the Unix domain socket server. *)
 103   server := Some (
 104     Whenproto_srv.When.V1.create_server
 105       ~proc_reload_file
 106       ~proc_set_variable
 107       ~proc_get_variable
 108       ~proc_get_variable_names
 109       ~proc_exit_daemon
 110       ~proc_get_jobs
 111       ~proc_cancel_job
 112       ~proc_start_job
 113       ~proc_get_job
 114       ~proc_set_variables
 115       ~proc_get_job_names
 116       ~proc_test_variables
 117       ~proc_ping_daemon
 118       ~proc_whisper_variables
 119       (Rpc_server.Unix addr)
 120       Rpc.Tcp (* not TCP, this is the same as SOCK_STREAM *)
 121       Rpc.Socket
 122       esys
 123   );
 124
 125   (* Handle SIGCHLD to clean up jobs. *)
 126   Sys.set_signal Sys.sigchld (Sys.Signal_handle handle_sigchld);
 127
 128   (* Load or initialize the variables. *)
 129   let variables_file = sprintf "%s/variables" !jobsdir in
 130   state :=
 131     try
 132       let chan = open_in variables_file in
 133       let r = load_variables !state chan in
 134       close_in chan;
 135       r
 136     with
 137     | Sys_error _ ->
 138       Whenstate.set_variable !state "JOBSERIAL" (T_int zero_big_int)
 139
 140 (* Try to load the variables from the file.  If the file exists and
 141  * cannot be read, raise an exception.
 142  *)
 143 and load_variables state chan =
 144   let signature = input_line chan in
 145   if signature = "WHENJOBS VARIABLES VERSION 1" then (
 146     let variables : variables_file_v1 = input_value chan in
 147     List.fold_left (
 148       fun state (n, v) ->
 149         Whenstate.set_variable state n (variable_of_variable_v1 v)
 150     ) state variables
 151   ) else (* in future, other signatures, but for now ... *)
 152     failwith (sprintf "cannot read variables file: invalid signature: %s"
 153                 signature)
 154
 155 and save_variables () =
 156   let variables_file = sprintf "%s/variables" !jobsdir in
 157   let new_file = variables_file ^ ".new" in
 158   let chan = open_out new_file in
 159   fprintf chan "WHENJOBS VARIABLES VERSION 1\n";
 160   let variables = Whenstate.get_variables !state in
 161   let variables =
 162     List.map (fun (n, v) -> n, variable_v1_of_variable v) variables in
 163   output_value chan variables;
 164
 165   (* Try to arrange that the new file is updated atomically. *)
 166   flush chan;
 167   Netsys_posix.fsync (descr_of_out_channel chan);
 168   close_out chan;
 169   rename new_file variables_file
 170
 171 and proc_reload_file () =
 172   if !debug then Syslog.notice "remote call: reload_file";
 173
 174   try reload_files (); `ok
 175   with Failure err -> `error err
 176
 177 and proc_set_variable (name, value) =
 178   if !debug then Syslog.notice "remote call: set_variable %s" name;
 179
 180   try
 181     check_valid_variable_name name;
 182
 183     let value = variable_of_rpc value in
 184     state := Whenstate.set_variable !state name value;
 185
 186     (* Which jobs need to be re-evaluated? *)
 187     let jobs = Whenstate.get_dependencies !state [name] in
 188     let jobnames, state' = reevaluate_whenjobs !state jobs in
 189     let state' = run_whenjobs state' jobnames in
 190     state := state';
 191     save_variables ();
 192
 193     `ok
 194   with
 195     Failure msg -> `error msg
 196
 197 and proc_get_variable name =
 198   if !debug then Syslog.notice "remote call: get_variable %s" name;
 199
 200   rpc_of_variable (Whenstate.get_variable !state name)
 201
 202 and proc_get_variable_names () =
 203   if !debug then Syslog.notice "remote call: get_variable_names";
 204
 205   let vars = Whenstate.get_variable_names !state in
 206
 207   (* Return variable names as a sorted array. *)
 208   let vars = Array.of_list vars in
 209   Array.sort compare vars;
 210   vars
 211
 212 and proc_exit_daemon () =
 213   if !debug then Syslog.notice "remote call: exit_daemon";
 214
 215   match !server with
 216   | None ->
 217     `error "exit_daemon: no server handle"
 218   | Some s ->
 219     Rpc_server.stop_server ~graceful:true s;
 220     server := None;
 221     Gc.compact (); (* force the server handle to get cleaned up now *)
 222     `ok
 223
 224 and proc_get_jobs () =
 225   let running = Array.of_list (IntMap.values !runningmap) in
 226   Array.map (
 227     fun (job, dir, serial, start_time) ->
 228       { Whenproto_aux.job_name = job.job_name;
 229         job_serial = string_of_big_int serial;
 230         job_tmpdir = dir; job_start_time = Int64.of_float start_time }
 231   ) running
 232
 233 and proc_cancel_job serial =
 234   try
 235     let serial = big_int_of_string serial in
 236     let pid = BigIntMap.find serial !serialmap in
 237     kill pid 15;
 238     `ok
 239   with
 240   | Not_found -> `error "job not found"
 241   | exn -> `error (Printexc.to_string exn)
 242
 243 and proc_start_job jobname =
 244   try
 245     let job = Whenstate.get_job !state jobname in
 246     let state' = run_job !state job in
 247     state := state';
 248     save_variables ();
 249     `ok
 250   with
 251   | Not_found -> `error "job not found"
 252   | exn -> `error (Printexc.to_string exn)
 253
 254 and proc_get_job serial =
 255   try
 256     let serial = big_int_of_string serial in
 257     let pid = BigIntMap.find serial !serialmap in
 258     let job, dir, serial, start_time = IntMap.find pid !runningmap in
 259     { Whenproto_aux.job_name = job.job_name;
 260       job_serial = string_of_big_int serial;
 261       job_tmpdir = dir; job_start_time = Int64.of_float start_time }
 262   with
 263   | Not_found -> failwith "job not found"
 264   | exn -> failwith (Printexc.to_string exn)
 265
 266 and proc_set_variables vars =
 267   try
 268     let vars = Array.map (
 269       fun { Whenproto_aux.sv_name = name; sv_value = value } ->
 270         name, variable_of_rpc value
 271     ) vars in
 272     let vars = Array.to_list vars in
 273
 274     if !debug then
 275       Syslog.notice "remote call: set_variables (%s)"
 276         (String.concat " "
 277            (List.map (
 278              fun (name, value) ->
 279                sprintf "%s=%s" name (string_of_variable value)
 280             ) vars));
 281
 282     List.iter (fun (name, _) -> check_valid_variable_name name) vars;
 283
 284     (* Update all the variables atomically. *)
 285     let s = List.fold_left (
 286       fun s (name, value) -> Whenstate.set_variable s name value
 287     ) !state vars in
 288     state := s;
 289
 290     (* Which jobs need to be re-evaluated? *)
 291     let jobs = Whenstate.get_dependencies !state (List.map fst vars) in
 292     let jobnames, state' = reevaluate_whenjobs !state jobs in
 293     let state' = run_whenjobs state' jobnames in
 294     state := state';
 295     save_variables ();
 296
 297     `ok
 298   with
 299     Failure msg -> `error msg
 300
 301 and proc_get_job_names () =
 302   Array.of_list (Whenstate.get_job_names !state)
 303
 304 and proc_test_variables vars =
 305   (* This is the same as proc_set_variables, except that it doesn't
 306    * update the state, it just returns the jobs that *would* run if
 307    * these variables were set to these values.
 308    *)
 309   let vars = Array.map (
 310     fun { Whenproto_aux.sv_name = name; sv_value = value } ->
 311       name, variable_of_rpc value
 312   ) vars in
 313   let vars = Array.to_list vars in
 314
 315   if !debug then
 316     Syslog.notice "remote call: test_variables (%s)"
 317       (String.concat " "
 318          (List.map (
 319            fun (name, value) ->
 320              sprintf "%s=%s" name (string_of_variable value)
 321           ) vars));
 322
 323   List.iter (fun (name, _) -> check_valid_variable_name name) vars;
 324
 325   (* Update all the variables atomically. *)
 326   let state = List.fold_left (
 327     fun s (name, value) -> Whenstate.set_variable s name value
 328   ) !state vars in
 329
 330   (* Which jobs WOULD be re-evaluated? *)
 331   let jobs = Whenstate.get_dependencies state (List.map fst vars) in
 332   let jobnames, _ = reevaluate_whenjobs state jobs in
 333
 334   (* Return the names. *)
 335   Array.of_list jobnames
 336
 337 and proc_ping_daemon () = `ok
 338
 339 and proc_whisper_variables vars =
 340   try
 341     let vars = Array.map (
 342       fun { Whenproto_aux.sv_name = name; sv_value = value } ->
 343         name, variable_of_rpc value
 344     ) vars in
 345     let vars = Array.to_list vars in
 346
 347     if !debug then
 348       Syslog.notice "remote call: whisper_variables (%s)"
 349         (String.concat " "
 350            (List.map (
 351              fun (name, value) ->
 352                sprintf "%s=%s" name (string_of_variable value)
 353             ) vars));
 354
 355     List.iter (fun (name, _) -> check_valid_variable_name name) vars;
 356
 357     (* Update all the variables atomically. *)
 358     let s = List.fold_left (
 359       fun s (name, value) -> Whenstate.set_variable s name value
 360     ) !state vars in
 361     state := s;
 362     save_variables ();
 363
 364     (* .. but don't reevaluate or run jobs. *)
 365
 366     `ok
 367   with
 368     Failure msg -> `error msg
 369
 370 (* Reload the jobs file(s). *)
 371 and reload_files () =
 372   (* Get the highest numbered dir/jobs__*.cmo (bytecode) or
 373    * dir/jobs__*.cmxs (native code) file and load it.  Delete
 374    * lower-numbered (== older) files.
 375    *)
 376   let filename =
 377     let suffix, slen =
 378       if not Dynlink.is_native then ".cmo", 4 else ".cmxs", 5 in
 379     let dir = !jobsdir in
 380     let files = Array.to_list (Sys.readdir dir) in
 381     let times = filter_map (
 382       fun file ->
 383         if not (string_startswith file "jobs__") ||
 384           not (string_endswith file suffix) then
 385           None
 386         else (
 387           let len = String.length file in
 388           let t = String.sub file 6 (len-slen-6) in
 389           (* Use int64 because t won't necessarily fit into 31 bit int. *)
 390           try Some (Int64.of_string t)
 391           with Failure "int_of_string" -> assert false
 392         )
 393     ) files in
 394     let times = List.rev (List.sort compare times) in
 395     match times with
 396     | [] -> None
 397     | x::xs ->
 398       (* Unlink the older files. *)
 399       List.iter (
 400         fun t ->
 401           try unlink (dir // sprintf "jobs__%Ld%s" t suffix)
 402           with Unix_error _ -> ()
 403       ) xs;
 404       (* Return the newest (highest numbered) file. *)
 405       Some (dir // sprintf "jobs__%Ld%s" x suffix) in
 406
 407   (* As we are reloading the file, we want to create a new state
 408    * that has no jobs, but has all the variables from the previous
 409    * state.
 410    *)
 411   let s = Whenstate.copy_variables !state Whenstate.empty in
 412   Whenfile.init s;
 413
 414   let s =
 415     match filename with
 416     | None ->
 417       (* no jobs file, return the same state *)
 418       Syslog.notice "no jobs file found";
 419       s
 420     | Some filename ->
 421       try
 422         Dynlink.loadfile filename;
 423         let s = Whenfile.get_state () in
 424         Syslog.notice "loaded %d job(s)" (Whenstate.nr_jobs s);
 425         s
 426       with
 427       | Dynlink.Error err ->
 428         let err = Dynlink.error_message err in
 429         Syslog.error "error loading jobs: %s" err;
 430         failwith err
 431       | exn ->
 432         failwith (Printexc.to_string exn) in
 433
 434   let s = Whenstate.copy_prev_state !state s in
 435   state := s;
 436
 437   (* Re-evaluate all when jobs. *)
 438   let jobs = Whenstate.get_whenjobs !state in
 439   let jobnames, state' = reevaluate_whenjobs ~onload:true !state jobs in
 440   let state' = run_whenjobs state' jobnames in
 441   state := state';
 442   save_variables ();
 443
 444   (* Schedule the next every job to run. *)
 445   schedule_next_everyjob ()
 446
 447 (* Re-evaluate each when-statement job, in a loop until we reach
 448  * a fixpoint.  Return the list of job names that should run and
 449  * the updated state.
 450  *)
 451 and reevaluate_whenjobs ?onload state jobs =
 452   let rec loop (set, state) jobs =
 453     let set', state' =
 454       List.fold_left (
 455         fun (set, state) job ->
 456           let r, state' =
 457             try Whenstate.evaluate_whenjob ?onload state job
 458             with Invalid_argument err | Failure err ->
 459               Syslog.error "error evaluating job %s (at %s): %s"
 460                 job.job_name (Camlp4.PreCast.Ast.Loc.to_string job.job_loc) err;
 461               false, state in
 462
 463           if !debug then
 464             Syslog.notice "evaluate %s -> %b\n" job.job_name r;
 465
 466           (if r then StringSet.add job.job_name set else set), state'
 467       ) (set, state) jobs in
 468     (* reached a fixpoint? *)
 469     if StringSet.compare set set' <> 0 then
 470       loop (set', state') jobs
 471     else
 472       (set', state')
 473   in
 474   let set, state = loop (StringSet.empty, state) jobs in
 475   let jobnames = StringSet.elements set in
 476
 477   (* Ensure the jobs always run in predictable (name) order. *)
 478   let jobnames = List.sort compare_jobnames jobnames in
 479   jobnames, state
 480
 481 and run_whenjobs state jobnames =
 482   (* Run the jobs. *)
 483   let jobs = List.map (Whenstate.get_job state) jobnames in
 484   List.fold_left run_job state jobs
 485
 486 (* Schedule the next every-statement job to run, if there is one.  We
 487  * look at the every jobs, work out the time that each must run at,
 488  * pick the job(s) which must run soonest, and schedule a timer to run
 489  * them.  When the timer fires, it runs those jobs, then calls this
 490  * function again.
 491  *)
 492 and schedule_next_everyjob () =
 493   let t = time () in
 494
 495   (* Get only everyjobs. *)
 496   let jobs = Whenstate.get_everyjobs !state in
 497   let jobs = List.map (
 498     function
 499     | { job_cond = Every_job period } as job -> (job, period)
 500     | { job_cond = When_job _ } -> assert false
 501   ) jobs in
 502
 503   (* Map everyjob to next time it must run. *)
 504   let jobs = List.map (
 505     fun (job, period) ->
 506       let t' = next_periodexpr t period in
 507       assert (t' > t); (* serious bug in next_periodexpr if false *)
 508       job, t'
 509   ) jobs in
 510
 511   (* Sort, soonest first. *)
 512   let jobs = List.sort (fun (_,a) (_,b) -> compare a b) jobs in
 513
 514   if !debug then (
 515     List.iter (
 516       fun (job, t) ->
 517         Syslog.notice "%s: next scheduled run at %s"
 518           job.job_name (string_of_time_t t)
 519     ) jobs
 520   );
 521
 522   (* Pick the job(s) which run soonest. *)
 523   let rec pick = function
 524     | [] -> 0., []
 525     | [j, t] -> t, [j]
 526     | (j1, t) :: (j2, t') :: _ when t < t' -> t, [j1]
 527     | (j1, t) :: (((j2, t') :: _) as rest) -> t, (j1 :: snd (pick rest))
 528   in
 529   let t, jobs = pick jobs in
 530
 531   if t > 0. then (
 532     if jobs <> [] then (
 533       (* Ensure the jobs always run in predictable (name) order. *)
 534       let jobs =
 535         List.sort (fun {job_name = a} {job_name = b} -> compare_jobnames a b)
 536           jobs in
 537
 538       if !debug then
 539         Syslog.notice "scheduling job(s) %s to run at %s"
 540           (String.concat ", " (List.map (fun { job_name = name } -> name) jobs))
 541           (string_of_time_t t);
 542
 543       (* Schedule them to run at time t. *)
 544       let g = new_timer_group () in
 545       let t_diff = t -. Unix.time () in
 546       let t_diff = if t_diff < 0. then 0. else t_diff in
 547       let run_jobs () =
 548         delete_timer_group ();          (* Delete the timer. *)
 549         let state' = List.fold_left run_job !state jobs in
 550         state := state';
 551         save_variables ();
 552         schedule_next_everyjob ()
 553       in
 554       Unixqueue.weak_once esys g t_diff run_jobs;
 555     )
 556   )
 557
 558 and new_timer_group () =
 559   delete_timer_group ();
 560   let g = Unixqueue.new_group esys in
 561   timer_group := Some g;
 562   g
 563
 564 and delete_timer_group () =
 565   match !timer_group with
 566   | None -> ()
 567   | Some g ->
 568     Unixqueue.clear esys g;
 569     timer_group := None
 570
 571 and run_job state job =
 572   (* Increment JOBSERIAL. *)
 573   let serial, state =
 574     match Whenstate.get_variable state "JOBSERIAL" with
 575     | T_int serial ->
 576       let serial = succ_big_int serial in
 577       let state' = Whenstate.set_variable state "JOBSERIAL" (T_int serial) in
 578       serial, state'
 579     | _ -> assert false in
 580
 581   (* Call the pre-condition script.  Note this may decide not to run
 582    * the job by returning false.
 583    *)
 584   let pre_condition () =
 585     match job.job_pre with
 586     | None -> true
 587     | Some pre ->
 588       let rs = ref [] in
 589       IntMap.iter (
 590         fun pid (job, _, serial, start_time) ->
 591           let r = { pirun_job_name = job.job_name;
 592                     pirun_serial = serial;
 593                     pirun_start_time = start_time;
 594                     pirun_pid = pid } in
 595           rs := r :: !rs
 596       ) !runningmap;
 597       let preinfo = {
 598         pi_job_name = job.job_name;
 599         pi_serial = serial;
 600         pi_variables = Whenstate.get_variables state;
 601         pi_running = !rs;
 602       } in
 603       pre preinfo
 604   in
 605   if pre_condition () then (
 606     Syslog.notice "running %s (JOBSERIAL=%s)"
 607       job.job_name (string_of_big_int serial);
 608
 609     (* Create a temporary directory.  The current directory of the job
 610      * will be in this directory.  The directory is removed when the
 611      * child process exits.
 612      *)
 613     let dir = tmpdir () in
 614
 615     let pid = fork () in
 616     if pid = 0 then ( (* child process running the job *)
 617       chdir dir;
 618
 619       (* Set environment variables corresponding to each variable. *)
 620       List.iter
 621         (fun (name, value) -> putenv name (string_of_variable value))
 622         (Whenstate.get_variables state);
 623
 624       (* Set the $JOBNAME environment variable. *)
 625       putenv "JOBNAME" job.job_name;
 626
 627       (* Create a temporary file containing the shell script fragment. *)
 628       let script = dir // "script.sh" in
 629       let chan = open_out script in
 630       fprintf chan "set -e\n"; (* So that jobs exit on error. *)
 631       output_string chan job.job_script.sh_script;
 632       close_out chan;
 633       chmod script 0o700;
 634
 635       let shell = try getenv "SHELL" with Not_found -> "/bin/sh" in
 636
 637       (* Set output to file. *)
 638       let output = dir // "output.txt" in
 639       let fd = openfile output [O_WRONLY; O_CREAT; O_TRUNC; O_NOCTTY] 0o600 in
 640       dup2 fd stdout;
 641       dup2 fd stderr;
 642       close fd;
 643
 644       (* Execute the shell script. *)
 645       (try execvp shell [| shell; "-c"; script |];
 646        with Unix_error (err, fn, _) ->
 647          Syslog.error "%s failed: %s: %s" fn script (error_message err)
 648       );
 649       _exit 1
 650     );
 651
 652     (* Remember this PID, the job and the temporary directory, so we
 653      * can clean up when the child exits.
 654      *)
 655     runningmap := IntMap.add pid (job, dir, serial, time ()) !runningmap;
 656     serialmap := BigIntMap.add serial pid !serialmap;
 657
 658     state
 659   )
 660   else (
 661     Syslog.notice "not running %s (JOBSERIAL=%s) because pre() condition returned false"
 662       job.job_name (string_of_big_int serial);
 663
 664     state
 665   )
 666
 667 and tmpdir () =
 668   let chan = open_in "/dev/urandom" in
 669   let data = String.create 16 in
 670   really_input chan data 0 (String.length data);
 671   close_in chan;
 672   let data = Digest.to_hex (Digest.string data) in
 673   let dir = Filename.temp_dir_name // sprintf "whenjobs%s" data in
 674   mkdir dir 0o700;
 675   dir
 676
 677 (* This is called when a job (child process) exits. *)
 678 and handle_sigchld _ =
 679   try
 680     let pid, status = waitpid [WNOHANG] 0 in
 681     if pid > 0 then (
 682       (* Look up the PID in the running jobs map. *)
 683       let job, dir, serial, time = IntMap.find pid !runningmap in
 684       runningmap := IntMap.remove pid !runningmap;
 685       serialmap := BigIntMap.remove serial !serialmap;
 686       post_job job dir serial time status
 687     )
 688   with Unix_error _ | Not_found -> ()
 689
 690 and post_job job dir serial time status =
 691   (* If there is a post function, run it. *)
 692   (match job.job_post with
 693   | None -> ()
 694   | Some post ->
 695     let code =
 696       match status with
 697       | WEXITED c -> c
 698       | WSIGNALED s | WSTOPPED s -> 1 in
 699     let result = {
 700       res_job_name = job.job_name;
 701       res_serial = serial;
 702       res_code = code;
 703       res_tmpdir = dir;
 704       res_output = dir // "output.txt";
 705       res_start_time = time
 706     } in
 707     try post result
 708     with
 709     | Failure msg ->
 710       Syslog.error "job %s post function failed: %s" job.job_name msg
 711     | exn ->
 712       Syslog.error "job %s post function exception: %s"
 713         job.job_name (Printexc.to_string exn)
 714   );
 715
 716   (* This should be safe because the path cannot contain shell metachars. *)
 717   let cmd = sprintf "rm -rf '%s'" dir in
 718   ignore (Sys.command cmd)
 719
 720 (* Intelligent comparison of job names. *)
 721 and compare_jobnames name1 name2 =
 722   try
 723     let len1 = String.length name1
 724     and len2 = String.length name2 in
 725     if len1 > 4 && len2 > 4 &&
 726       String.sub name1 0 4 = "job$" && String.sub name2 0 4 = "job$"
 727     then (
 728       let i1 = int_of_string (String.sub name1 4 (len1-4)) in
 729       let i2 = int_of_string (String.sub name2 4 (len2-4)) in
 730       compare i1 i2
 731     )
 732     else raise Not_found
 733   with _ ->
 734     compare name1 name2
 735
 736 let main_loop () =
 737   Unixqueue.run esys