git.sur5r.net Git - bacula/bacula/blob - bacula/src/dird/scheduler.c

   1 /*
   2    Bacula® - The Network Backup Solution
   3
   4    Copyright (C) 2000-2009 Free Software Foundation Europe e.V.
   5
   6    The main author of Bacula is Kern Sibbald, with contributions from
   7    many others, a complete list can be found in the file AUTHORS.
   8    This program is Free Software; you can redistribute it and/or
   9    modify it under the terms of version two of the GNU General Public
  10    License as published by the Free Software Foundation and included
  11    in the file LICENSE.
  12
  13    This program is distributed in the hope that it will be useful, but
  14    WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16    General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  21    02110-1301, USA.
  22
  23    Bacula® is a registered trademark of Kern Sibbald.
  24    The licensor of Bacula is the Free Software Foundation Europe
  25    (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich,
  26    Switzerland, email:ftf@fsfeurope.org.
  27 */
  28 /*
  29  *
  30  *   Bacula scheduler
  31  *     It looks at what jobs are to be run and when
  32  *     and waits around until it is time to
  33  *     fire them up.
  34  *
  35  *     Kern Sibbald, May MM, major revision December MMIII
  36  *
  37  *   Version $Id$
  38  */
  39
  40 #include "bacula.h"
  41 #include "dird.h"
  42
  43 #if 0
  44 #define SCHED_DEBUG
  45 #define DBGLVL 0
  46 #else
  47 #undef SCHED_DEBUG
  48 #define DBGLVL 200
  49 #endif
  50
  51 const int dbglvl = DBGLVL;
  52
  53 /* Local variables */
  54 struct job_item {
  55    RUN *run;
  56    JOB *job;
  57    time_t runtime;
  58    int Priority;
  59    dlink link;                        /* link for list */
  60 };
  61
  62 /* List of jobs to be run. They were scheduled in this hour or the next */
  63 static dlist *jobs_to_run;               /* list of jobs to be run */
  64
  65 /* Time interval in secs to sleep if nothing to be run */
  66 static int const next_check_secs = 60;
  67
  68 /* Forward referenced subroutines */
  69 static void find_runs();
  70 static void add_job(JOB *job, RUN *run, time_t now, time_t runtime);
  71 static void dump_job(job_item *ji, const char *msg);
  72
  73 /* Imported subroutines */
  74
  75 /* Imported variables */
  76
  77 /**
  78  * called by reload_config to tell us that the schedules
  79  * we may have based our next jobs to run queues have been
  80  * invalidated.  In fact the schedules may not have changed
  81  * but the run object that we have recorded the last_run time
  82  * on are new and no longer have a valid last_run time which
  83  * causes us to double run schedules that get put into the list
  84  * because run_nh = 1.
  85  */
  86 static bool schedules_invalidated = false;
  87 void invalidate_schedules(void) {
  88     schedules_invalidated = true;
  89 }
  90
  91 /*********************************************************************
  92  *
  93  *         Main Bacula Scheduler
  94  *
  95  */
  96 JCR *wait_for_next_job(char *one_shot_job_to_run)
  97 {
  98    JCR *jcr;
  99    JOB *job;
 100    RUN *run;
 101    time_t now, prev;
 102    static bool first = true;
 103    job_item *next_job = NULL;
 104
 105    Dmsg0(dbglvl, "Enter wait_for_next_job\n");
 106    if (first) {
 107       first = false;
 108       /* Create scheduled jobs list */
 109       jobs_to_run = New(dlist(next_job, &next_job->link));
 110       if (one_shot_job_to_run) {            /* one shot */
 111          job = (JOB *)GetResWithName(R_JOB, one_shot_job_to_run);
 112          if (!job) {
 113             Emsg1(M_ABORT, 0, _("Job %s not found\n"), one_shot_job_to_run);
 114          }
 115          Dmsg1(5, "Found one_shot_job_to_run %s\n", one_shot_job_to_run);
 116          jcr = new_jcr(sizeof(JCR), dird_free_jcr);
 117          set_jcr_defaults(jcr, job);
 118          return jcr;
 119       }
 120    }
 121
 122    /* Wait until we have something in the
 123     * next hour or so.
 124     */
 125 again:
 126    while (jobs_to_run->empty()) {
 127       find_runs();
 128       if (!jobs_to_run->empty()) {
 129          break;
 130       }
 131       bmicrosleep(next_check_secs, 0); /* recheck once per minute */
 132    }
 133
 134 #ifdef  list_chain
 135    job_item *je;
 136    foreach_dlist(je, jobs_to_run) {
 137       dump_job(je, _("Walk queue"));
 138    }
 139 #endif
 140    /*
 141     * Pull the first job to run (already sorted by runtime and
 142     *  Priority, then wait around until it is time to run it.
 143     */
 144    next_job = (job_item *)jobs_to_run->first();
 145    jobs_to_run->remove(next_job);
 146
 147    dump_job(next_job, _("Dequeued job"));
 148
 149    if (!next_job) {                /* we really should have something now */
 150       Emsg0(M_ABORT, 0, _("Scheduler logic error\n"));
 151    }
 152
 153    /* Now wait for the time to run the job */
 154    for (;;) {
 155       time_t twait;
 156       /** discard scheduled queue and rebuild with new schedule objects. **/
 157       lock_jobs();
 158       if (schedules_invalidated) {
 159           dump_job(next_job, "Invalidated job");
 160           free(next_job);
 161           while (!jobs_to_run->empty()) {
 162               next_job = (job_item *)jobs_to_run->first();
 163               jobs_to_run->remove(next_job);
 164               dump_job(next_job, "Invalidated job");
 165               free(next_job);
 166           }
 167           schedules_invalidated = false;
 168           unlock_jobs();
 169           goto again;
 170       }
 171       unlock_jobs();
 172       prev = now = time(NULL);
 173       twait = next_job->runtime - now;
 174       if (twait <= 0) {               /* time to run it */
 175          break;
 176       }
 177       /* Recheck at least once per minute */
 178       bmicrosleep((next_check_secs < twait)?next_check_secs:twait, 0);
 179       /* Attempt to handle clock shift (but not daylight savings time changes)
 180        * we allow a skew of 10 seconds before invalidating everything.
 181        */
 182       now = time(NULL);
 183       if (now < prev-10 || now > (prev+next_check_secs+10)) {
 184          schedules_invalidated = true;
 185       }
 186    }
 187    jcr = new_jcr(sizeof(JCR), dird_free_jcr);
 188    run = next_job->run;               /* pick up needed values */
 189    job = next_job->job;
 190    if (job->enabled) {
 191       dump_job(next_job, _("Run job"));
 192    }
 193    free(next_job);
 194    if (!job->enabled) {
 195       free_jcr(jcr);
 196       goto again;                     /* ignore this job */
 197    }
 198    run->last_run = now;               /* mark as run now */
 199
 200    ASSERT(job);
 201    set_jcr_defaults(jcr, job);
 202    if (run->level) {
 203       jcr->set_JobLevel(run->level);  /* override run level */
 204    }
 205    if (run->pool) {
 206       jcr->pool = run->pool;          /* override pool */
 207       jcr->run_pool_override = true;
 208    }
 209    if (run->full_pool) {
 210       jcr->full_pool = run->full_pool; /* override full pool */
 211       jcr->run_full_pool_override = true;
 212    }
 213    if (run->inc_pool) {
 214       jcr->inc_pool = run->inc_pool;  /* override inc pool */
 215       jcr->run_inc_pool_override = true;
 216    }
 217    if (run->diff_pool) {
 218       jcr->diff_pool = run->diff_pool;  /* override dif pool */
 219       jcr->run_diff_pool_override = true;
 220    }
 221    if (run->storage) {
 222       USTORE store;
 223       store.store = run->storage;
 224       pm_strcpy(store.store_source, _("run override"));
 225       set_rwstorage(jcr, &store);     /* override storage */
 226    }
 227    if (run->msgs) {
 228       jcr->messages = run->msgs;      /* override messages */
 229    }
 230    if (run->Priority) {
 231       jcr->JobPriority = run->Priority;
 232    }
 233    if (run->spool_data_set) {
 234       jcr->spool_data = run->spool_data;
 235    }
 236    if (run->write_part_after_job_set) {
 237       jcr->write_part_after_job = run->write_part_after_job;
 238    }
 239    Dmsg0(dbglvl, "Leave wait_for_next_job()\n");
 240    return jcr;
 241 }
 242
 243
 244 /*
 245  * Shutdown the scheduler
 246  */
 247 void term_scheduler()
 248 {
 249    if (jobs_to_run) {
 250       delete jobs_to_run;
 251    }
 252 }
 253
 254 /*
 255  * Find all jobs to be run this hour and the next hour.
 256  */
 257 static void find_runs()
 258 {
 259    time_t now, next_hour, runtime;
 260    RUN *run;
 261    JOB *job;
 262    SCHED *sched;
 263    struct tm tm;
 264    int minute;
 265    int hour, mday, wday, month, wom, woy;
 266    /* Items corresponding to above at the next hour */
 267    int nh_hour, nh_mday, nh_wday, nh_month, nh_wom, nh_woy, nh_year;
 268
 269    Dmsg0(dbglvl, "enter find_runs()\n");
 270
 271
 272    /* compute values for time now */
 273    now = time(NULL);
 274    (void)localtime_r(&now, &tm);
 275    hour = tm.tm_hour;
 276    minute = tm.tm_min;
 277    mday = tm.tm_mday - 1;
 278    wday = tm.tm_wday;
 279    month = tm.tm_mon;
 280    wom = mday / 7;
 281    woy = tm_woy(now);                     /* get week of year */
 282
 283    Dmsg7(dbglvl, "now = %x: h=%d m=%d md=%d wd=%d wom=%d woy=%d\n",
 284          now, hour, month, mday, wday, wom, woy);
 285
 286    /*
 287     * Compute values for next hour from now.
 288     * We do this to be sure we don't miss a job while
 289     * sleeping.
 290     */
 291    next_hour = now + 3600;
 292    (void)localtime_r(&next_hour, &tm);
 293    nh_hour = tm.tm_hour;
 294    nh_mday = tm.tm_mday - 1;
 295    nh_wday = tm.tm_wday;
 296    nh_month = tm.tm_mon;
 297    nh_year  = tm.tm_year;
 298    nh_wom = nh_mday / 7;
 299    nh_woy = tm_woy(now);                     /* get week of year */
 300
 301    Dmsg7(dbglvl, "nh = %x: h=%d m=%d md=%d wd=%d wom=%d woy=%d\n",
 302          next_hour, nh_hour, nh_month, nh_mday, nh_wday, nh_wom, nh_woy);
 303
 304    /* Loop through all jobs */
 305    LockRes();
 306    foreach_res(job, R_JOB) {
 307       sched = job->schedule;
 308       if (sched == NULL || !job->enabled) { /* scheduled? or enabled? */
 309          continue;                    /* no, skip this job */
 310       }
 311       Dmsg1(dbglvl, "Got job: %s\n", job->hdr.name);
 312       for (run=sched->run; run; run=run->next) {
 313          bool run_now, run_nh;
 314          /*
 315           * Find runs scheduled between now and the next hour.
 316           */
 317 #ifdef xxxx
 318          Dmsg0(000, "\n");
 319          Dmsg6(000, "run h=%d m=%d md=%d wd=%d wom=%d woy=%d\n",
 320             hour, month, mday, wday, wom, woy);
 321          Dmsg6(000, "bitset bsh=%d bsm=%d bsmd=%d bswd=%d bswom=%d bswoy=%d\n",
 322             bit_is_set(hour, run->hour),
 323             bit_is_set(month, run->month),
 324             bit_is_set(mday, run->mday),
 325             bit_is_set(wday, run->wday),
 326             bit_is_set(wom, run->wom),
 327             bit_is_set(woy, run->woy));
 328
 329          Dmsg6(000, "nh_run h=%d m=%d md=%d wd=%d wom=%d woy=%d\n",
 330             nh_hour, nh_month, nh_mday, nh_wday, nh_wom, nh_woy);
 331          Dmsg6(000, "nh_bitset bsh=%d bsm=%d bsmd=%d bswd=%d bswom=%d bswoy=%d\n",
 332             bit_is_set(nh_hour, run->hour),
 333             bit_is_set(nh_month, run->month),
 334             bit_is_set(nh_mday, run->mday),
 335             bit_is_set(nh_wday, run->wday),
 336             bit_is_set(nh_wom, run->wom),
 337             bit_is_set(nh_woy, run->woy));
 338 #endif
 339
 340          run_now = bit_is_set(hour, run->hour) &&
 341             bit_is_set(mday, run->mday) &&
 342             bit_is_set(wday, run->wday) &&
 343             bit_is_set(month, run->month) &&
 344             bit_is_set(wom, run->wom) &&
 345             bit_is_set(woy, run->woy);
 346
 347          run_nh = bit_is_set(nh_hour, run->hour) &&
 348             bit_is_set(nh_mday, run->mday) &&
 349             bit_is_set(nh_wday, run->wday) &&
 350             bit_is_set(nh_month, run->month) &&
 351             bit_is_set(nh_wom, run->wom) &&
 352             bit_is_set(nh_woy, run->woy);
 353
 354          Dmsg3(dbglvl, "run@%p: run_now=%d run_nh=%d\n", run, run_now, run_nh);
 355
 356          if (run_now || run_nh) {
 357            /* find time (time_t) job is to be run */
 358            (void)localtime_r(&now, &tm);      /* reset tm structure */
 359            tm.tm_min = run->minute;     /* set run minute */
 360            tm.tm_sec = 0;               /* zero secs */
 361            runtime = mktime(&tm);
 362            if (run_now) {
 363              add_job(job, run, now, runtime);
 364            }
 365            /* If job is to be run in the next hour schedule it */
 366            if (run_nh) {
 367              add_job(job, run, now, runtime + 3600);
 368            }
 369          }
 370       }
 371    }
 372    UnlockRes();
 373    Dmsg0(dbglvl, "Leave find_runs()\n");
 374 }
 375
 376 static void add_job(JOB *job, RUN *run, time_t now, time_t runtime)
 377 {
 378    job_item *ji;
 379    bool inserted = false;
 380    /*
 381     * Don't run any job that ran less than a minute ago, but
 382     *  do run any job scheduled less than a minute ago.
 383     */
 384    if (((runtime - run->last_run) < 61) || ((runtime+59) < now)) {
 385 #ifdef SCHED_DEBUG
 386       Dmsg4(000, "Drop: Job=\"%s\" run=%lld. last_run=%lld. now=%lld\n", job->hdr.name,
 387             (utime_t)runtime, (utime_t)run->last_run, (utime_t)now);
 388       fflush(stdout);
 389 #endif
 390       return;
 391    }
 392 #ifdef SCHED_DEBUG
 393    Dmsg4(000, "Add: Job=\"%s\" run=%lld last_run=%lld now=%lld\n", job->hdr.name,
 394             (utime_t)runtime, (utime_t)run->last_run, (utime_t)now);
 395 #endif
 396    /* accept to run this job */
 397    job_item *je = (job_item *)malloc(sizeof(job_item));
 398    je->run = run;
 399    je->job = job;
 400    je->runtime = runtime;
 401    if (run->Priority) {
 402       je->Priority = run->Priority;
 403    } else {
 404       je->Priority = job->Priority;
 405    }
 406
 407    /* Add this job to the wait queue in runtime, priority sorted order */
 408    foreach_dlist(ji, jobs_to_run) {
 409       if (ji->runtime > je->runtime ||
 410           (ji->runtime == je->runtime && ji->Priority > je->Priority)) {
 411          jobs_to_run->insert_before(je, ji);
 412          dump_job(je, _("Inserted job"));
 413          inserted = true;
 414          break;
 415       }
 416    }
 417    /* If place not found in queue, append it */
 418    if (!inserted) {
 419       jobs_to_run->append(je);
 420       dump_job(je, _("Appended job"));
 421    }
 422 #ifdef SCHED_DEBUG
 423    foreach_dlist(ji, jobs_to_run) {
 424       dump_job(ji, _("Run queue"));
 425    }
 426    Dmsg0(000, "End run queue\n");
 427 #endif
 428 }
 429
 430 static void dump_job(job_item *ji, const char *msg)
 431 {
 432 #ifdef SCHED_DEBUG
 433    char dt[MAX_TIME_LENGTH];
 434    int save_debug = debug_level;
 435    if (debug_level < dbglvl) {
 436       return;
 437    }
 438    bstrftime_nc(dt, sizeof(dt), ji->runtime);
 439    Dmsg4(dbglvl, "%s: Job=%s priority=%d run %s\n", msg, ji->job->hdr.name,
 440       ji->Priority, dt);
 441    fflush(stdout);
 442    debug_level = save_debug;
 443 #endif
 444 }