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

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