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

   1 /*
   2  *
   3  *   Bacula scheduler
   4  *     It looks at what jobs are to be run and when
   5  *     and waits around until it is time to
   6  *     fire them up.
   7  *
   8  *     Kern Sibbald, May MM, major revision December MMIII
   9  *
  10  *   Version $Id$
  11  */
  12 /*
  13    Copyright (C) 2000-2003 Kern Sibbald and John Walker
  14
  15    This program is free software; you can redistribute it and/or
  16    modify it under the terms of the GNU General Public License as
  17    published by the Free Software Foundation; either version 2 of
  18    the License, or (at your option) any later version.
  19
  20    This program is distributed in the hope that it will be useful,
  21    but WITHOUT ANY WARRANTY; without even the implied warranty of
  22    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  23    General Public License for more details.
  24
  25    You should have received a copy of the GNU General Public
  26    License along with this program; if not, write to the Free
  27    Software Foundation, Inc., 59 Temple Place - Suite 330, Boston,
  28    MA 02111-1307, USA.
  29
  30  */
  31
  32 #include "bacula.h"
  33 #include "dird.h"
  34
  35
  36 /* Local variables */
  37 struct job_item {
  38    RUN *run;
  39    JOB *job;
  40    time_t runtime;
  41    int Priority;
  42    dlink link;                        /* link for list */
  43 };
  44
  45 /* List of jobs to be run. They were scheduled in this hour or the next */
  46 static dlist *jobs_to_run;               /* list of jobs to be run */
  47
  48 /* Time interval in secs to sleep if nothing to be run */
  49 static int const NEXT_CHECK_SECS = 60;
  50
  51 /* Forward referenced subroutines */
  52 static void find_runs();
  53 static void add_job(JOB *job, RUN *run, time_t now, time_t runtime);
  54 static void dump_job(job_item *ji, char *msg);
  55
  56 /* Imported subroutines */
  57
  58 /* Imported variables */
  59
  60
  61 /*********************************************************************
  62  *
  63  *         Main Bacula Scheduler
  64  *
  65  */
  66 JCR *wait_for_next_job(char *one_shot_job_to_run)
  67 {
  68    JCR *jcr;
  69    JOB *job;
  70    RUN *run;
  71    time_t now;
  72    static bool first = true;
  73    job_item *next_job = NULL;
  74
  75    Dmsg0(200, "Enter wait_for_next_job\n");
  76    if (first) {
  77       first = false;
  78       /* Create scheduled jobs list */
  79       jobs_to_run = new dlist(next_job, &next_job->link);
  80       if (one_shot_job_to_run) {            /* one shot */
  81          job = (JOB *)GetResWithName(R_JOB, one_shot_job_to_run);
  82          if (!job) {
  83             Emsg1(M_ABORT, 0, _("Job %s not found\n"), one_shot_job_to_run);
  84          }
  85          Dmsg1(5, "Found one_shot_job_to_run %s\n", one_shot_job_to_run);
  86          jcr = new_jcr(sizeof(JCR), dird_free_jcr);
  87          set_jcr_defaults(jcr, job);
  88          return jcr;
  89       }
  90    }
  91    /* Wait until we have something in the
  92     * next hour or so.
  93     */
  94    while (jobs_to_run->empty()) {
  95       find_runs();
  96       if (!jobs_to_run->empty()) {
  97          break;
  98       }
  99       bmicrosleep(NEXT_CHECK_SECS, 0); /* recheck once per minute */
 100    }
 101
 102 #define list_chain
 103 #ifdef  list_chain
 104    for (job_item *je=NULL; (je=(job_item *)jobs_to_run->next(je)); ) {
 105       dump_job(je, "Walk queue");
 106    }
 107 #endif
 108    /*
 109     * Pull the first job to run (already sorted by runtime and
 110     *  Priority, then wait around until it is time to run it.
 111     */
 112    next_job = (job_item *)jobs_to_run->first();
 113    jobs_to_run->remove(next_job);
 114
 115    dump_job(next_job, "Dequeued job");
 116
 117    if (!next_job) {                /* we really should have something now */
 118       Emsg0(M_ABORT, 0, _("Scheduler logic error\n"));
 119    }
 120
 121    /* Now wait for the time to run the job */
 122    for (;;) {
 123       time_t twait;
 124       now = time(NULL);
 125       twait = next_job->runtime - now;
 126       if (twait <= 0) {               /* time to run it */
 127          break;
 128       }
 129       bmicrosleep(twait, 0);
 130    }
 131    run = next_job->run;               /* pick up needed values */
 132    job = next_job->job;
 133    run->last_run = now;               /* mark as run now */
 134
 135    dump_job(next_job, "Run job");
 136
 137    free(next_job);
 138
 139    jcr = new_jcr(sizeof(JCR), dird_free_jcr);
 140    ASSERT(job);
 141    set_jcr_defaults(jcr, job);
 142    if (run->level) {
 143       jcr->JobLevel = run->level;        /* override run level */
 144    }
 145    if (run->pool) {
 146       jcr->pool = run->pool;          /* override pool */
 147    }
 148    if (run->storage) {
 149       jcr->store = run->storage;      /* override storage */
 150    }
 151    if (run->msgs) {
 152       jcr->messages = run->msgs;      /* override messages */
 153    }
 154    if (run->Priority) {
 155       jcr->JobPriority = run->Priority;
 156    }
 157    Dmsg0(200, "Leave wait_for_next_job()\n");
 158    return jcr;
 159 }
 160
 161
 162 /*
 163  * Shutdown the scheduler
 164  */
 165 void term_scheduler()
 166 {
 167    /* Release all queued job entries to be run */
 168    for (void *je=NULL; (je=jobs_to_run->next(je)); ) {
 169       free(je);
 170    }
 171    delete jobs_to_run;
 172 }
 173
 174 /*
 175  * Find all jobs to be run this hour and the next hour.
 176  */
 177 static void find_runs()
 178 {
 179    time_t now, next_hour, runtime;
 180    RUN *run;
 181    JOB *job;
 182    SCHED *sched;
 183    struct tm tm;
 184    int minute;
 185    int hour, mday, wday, month, wom, woy;
 186    /* Items corresponding to above at the next hour */
 187    int nh_hour, nh_mday, nh_wday, nh_month, nh_wom, nh_woy, nh_year;
 188
 189    Dmsg0(200, "enter find_runs()\n");
 190
 191
 192    /* compute values for time now */
 193    now = time(NULL);
 194    localtime_r(&now, &tm);
 195    hour = tm.tm_hour;
 196    minute = tm.tm_min;
 197    mday = tm.tm_mday - 1;
 198    wday = tm.tm_wday;
 199    month = tm.tm_mon;
 200    wom = tm_wom(tm.tm_mday, tm.tm_wday);  /* get week of month */
 201    woy = tm_woy(now);                     /* get week of year */
 202
 203    /*
 204     * Compute values for next hour from now.
 205     * We do this to be sure we don't miss a job while
 206     * sleeping.
 207     */
 208    next_hour = now + 3600;
 209    localtime_r(&next_hour, &tm);
 210    nh_hour = tm.tm_hour;
 211    nh_mday = tm.tm_mday - 1;
 212    nh_wday = tm.tm_wday;
 213    nh_month = tm.tm_mon;
 214    nh_year  = tm.tm_year;
 215    nh_wom = tm_wom(tm.tm_mday, tm.tm_wday);  /* get week of month */
 216    nh_woy = tm_woy(now);                     /* get week of year */
 217
 218    /* Loop through all jobs */
 219    LockRes();
 220    for (job=NULL; (job=(JOB *)GetNextRes(R_JOB, (RES *)job)); ) {
 221       sched = job->schedule;
 222       if (sched == NULL) {            /* scheduled? */
 223          continue;                    /* no, skip this job */
 224       }
 225       Dmsg1(200, "Got job: %s\n", job->hdr.name);
 226       for (run=sched->run; run; run=run->next) {
 227          bool run_now, run_nh;
 228          /*
 229           * Find runs scheduled between now and the next hour.
 230           */
 231 #ifdef xxxx
 232          Dmsg0(000, "\n");
 233          Dmsg6(000, "run h=%d m=%d md=%d wd=%d wom=%d woy=%d\n",
 234             hour, month, mday, wday, wom, woy);
 235          Dmsg6(000, "bitset bsh=%d bsm=%d bsmd=%d bswd=%d bswom=%d bswoy=%d\n",
 236             bit_is_set(hour, run->hour),
 237             bit_is_set(month, run->month),
 238             bit_is_set(mday, run->mday),
 239             bit_is_set(wday, run->wday),
 240             bit_is_set(wom, run->wom),
 241             bit_is_set(woy, run->woy));
 242
 243          Dmsg6(000, "nh_run h=%d m=%d md=%d wd=%d wom=%d woy=%d\n",
 244             nh_hour, nh_month, nh_mday, nh_wday, nh_wom, nh_woy);
 245          Dmsg6(000, "nh_bitset bsh=%d bsm=%d bsmd=%d bswd=%d bswom=%d bswoy=%d\n",
 246             bit_is_set(nh_hour, run->hour),
 247             bit_is_set(nh_month, run->month),
 248             bit_is_set(nh_mday, run->mday),
 249             bit_is_set(nh_wday, run->wday),
 250             bit_is_set(nh_wom, run->wom),
 251             bit_is_set(nh_woy, run->woy));
 252 #endif
 253
 254          run_now = bit_is_set(hour, run->hour) &&
 255             (bit_is_set(mday, run->mday) || bit_is_set(wday, run->wday)) &&
 256             bit_is_set(month, run->month) &&
 257             bit_is_set(wom, run->wom) &&
 258             bit_is_set(woy, run->woy);
 259
 260          run_nh = bit_is_set(nh_hour, run->hour) &&
 261             (bit_is_set(nh_mday, run->mday) || bit_is_set(nh_wday, run->wday)) &&
 262             bit_is_set(nh_month, run->month) &&
 263             bit_is_set(nh_wom, run->wom) &&
 264             bit_is_set(nh_woy, run->woy);
 265
 266          Dmsg2(200, "run_now=%d run_nh=%d\n", run_now, run_nh);
 267
 268          /* find time (time_t) job is to be run */
 269          localtime_r(&now, &tm);      /* reset tm structure */
 270          tm.tm_min = run->minute;     /* set run minute */
 271          tm.tm_sec = 0;               /* zero secs */
 272          if (run_now) {
 273             runtime = mktime(&tm);
 274             add_job(job, run, now, runtime);
 275          }
 276          /* If job is to be run in the next hour schedule it */
 277          if (run_nh) {
 278             /* Set correct values */
 279             tm.tm_hour = nh_hour;
 280             tm.tm_mday = nh_mday + 1; /* fixup because we biased for tests above */
 281             tm.tm_mon = nh_month;
 282             tm.tm_year = nh_year;
 283             runtime = mktime(&tm);
 284             add_job(job, run, now, runtime);
 285          }
 286       }
 287    }
 288    UnlockRes();
 289    Dmsg0(200, "Leave find_runs()\n");
 290 }
 291
 292 static void add_job(JOB *job, RUN *run, time_t now, time_t runtime)
 293 {
 294    job_item *ji = NULL;
 295    bool inserted = false;
 296    /*
 297     * Don't run any job that ran less than a minute ago, but
 298     *  do run any job scheduled less than a minute ago.
 299     */
 300    if (((runtime - run->last_run) < 61) || ((runtime+59) < now)) {
 301 #ifdef xxx
 302       char dt[50], dt1[50];
 303       bstrftime(dt, sizeof(dt), runtime);
 304       strcpy(dt+7, dt+9);                /* cut century */
 305       bstrftime(dt1, sizeof(dt1), now);
 306       strcpy(dt1+7, dt1+9);                /* cut century */
 307       Dmsg2(000, "runtime=%s now=%s\n", dt, dt1);
 308 #endif
 309       return;
 310    }
 311    /* accept to run this job */
 312    job_item *je = (job_item *)malloc(sizeof(job_item));
 313    je->run = run;
 314    je->job = job;
 315    je->runtime = runtime;
 316    if (run->Priority) {
 317       je->Priority = run->Priority;
 318    } else {
 319       je->Priority = job->Priority;
 320    }
 321
 322    /* Add this job to the wait queue in runtime, priority sorted order */
 323    while ( (ji=(job_item *)jobs_to_run->next(ji)) ) {
 324       if (ji->runtime > je->runtime ||
 325           (ji->runtime == je->runtime && ji->Priority > je->Priority)) {
 326          jobs_to_run->insert_before(je, ji);
 327          dump_job(je, "Inserted job");
 328          inserted = true;
 329          break;
 330       }
 331    }
 332    /* If place not found in queue, append it */
 333    if (!inserted) {
 334       jobs_to_run->append(je);
 335       dump_job(je, "Appended job");
 336    }
 337 }
 338
 339 static void dump_job(job_item *ji, char *msg)
 340 {
 341    char dt[MAX_TIME_LENGTH];
 342    if (debug_level < 200) {
 343       return;
 344    }
 345    bstrftime(dt, sizeof(dt), ji->runtime);
 346    strcpy(dt+7, dt+9);                /* cut century */
 347    Dmsg4(200, "%s: Job=%s priority=%d run %s\n", msg, ji->job->hdr.name,
 348       ji->Priority, dt);
 349 }