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-2005 Kern Sibbald
  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
  17    version 2 as amended with additional clauses defined in the
  18    file LICENSE in the main source directory.
  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
  23    the file LICENSE for additional details.
  24
  25  */
  26
  27 #include "bacula.h"
  28 #include "dird.h"
  29
  30 /* #define SCHED_DEBUG */
  31
  32
  33 /* Local variables */
  34 struct job_item {
  35    RUN *run;
  36    JOB *job;
  37    time_t runtime;
  38    int Priority;
  39    dlink link;                        /* link for list */
  40 };
  41
  42 /* List of jobs to be run. They were scheduled in this hour or the next */
  43 static dlist *jobs_to_run;               /* list of jobs to be run */
  44
  45 /* Time interval in secs to sleep if nothing to be run */
  46 static int const NEXT_CHECK_SECS = 60;
  47
  48 /* Forward referenced subroutines */
  49 static void find_runs();
  50 static void add_job(JOB *job, RUN *run, time_t now, time_t runtime);
  51 static void dump_job(job_item *ji, const char *msg);
  52
  53 /* Imported subroutines */
  54
  55 /* Imported variables */
  56
  57
  58 /*********************************************************************
  59  *
  60  *         Main Bacula Scheduler
  61  *
  62  */
  63 JCR *wait_for_next_job(char *one_shot_job_to_run)
  64 {
  65    JCR *jcr;
  66    JOB *job;
  67    RUN *run;
  68    time_t now;
  69    static bool first = true;
  70    job_item *next_job = NULL;
  71
  72    Dmsg0(200, "Enter wait_for_next_job\n");
  73    if (first) {
  74       first = false;
  75       /* Create scheduled jobs list */
  76       jobs_to_run = New(dlist(next_job, &next_job->link));
  77       if (one_shot_job_to_run) {            /* one shot */
  78          job = (JOB *)GetResWithName(R_JOB, one_shot_job_to_run);
  79          if (!job) {
  80             Emsg1(M_ABORT, 0, _("Job %s not found\n"), one_shot_job_to_run);
  81          }
  82          Dmsg1(5, "Found one_shot_job_to_run %s\n", one_shot_job_to_run);
  83          jcr = new_jcr(sizeof(JCR), dird_free_jcr);
  84          set_jcr_defaults(jcr, job);
  85          return jcr;
  86       }
  87    }
  88    /* Wait until we have something in the
  89     * next hour or so.
  90     */
  91    while (jobs_to_run->empty()) {
  92       find_runs();
  93       if (!jobs_to_run->empty()) {
  94          break;
  95       }
  96       bmicrosleep(NEXT_CHECK_SECS, 0); /* recheck once per minute */
  97    }
  98
  99 #ifdef  list_chain
 100    job_item *je;
 101    foreach_dlist(je, jobs_to_run) {
 102       dump_job(je, _("Walk queue"));
 103    }
 104 #endif
 105    /*
 106     * Pull the first job to run (already sorted by runtime and
 107     *  Priority, then wait around until it is time to run it.
 108     */
 109    next_job = (job_item *)jobs_to_run->first();
 110    jobs_to_run->remove(next_job);
 111
 112    dump_job(next_job, _("Dequeued job"));
 113
 114    if (!next_job) {                /* we really should have something now */
 115       Emsg0(M_ABORT, 0, _("Scheduler logic error\n"));
 116    }
 117
 118    /* Now wait for the time to run the job */
 119    for (;;) {
 120       time_t twait;
 121       now = time(NULL);
 122       twait = next_job->runtime - now;
 123       if (twait <= 0) {               /* time to run it */
 124          break;
 125       }
 126       bmicrosleep(twait, 0);
 127    }
 128    run = next_job->run;               /* pick up needed values */
 129    job = next_job->job;
 130    run->last_run = now;               /* mark as run now */
 131
 132    dump_job(next_job, _("Run job"));
 133
 134    free(next_job);
 135
 136    jcr = new_jcr(sizeof(JCR), dird_free_jcr);
 137    ASSERT(job);
 138    set_jcr_defaults(jcr, job);
 139    if (run->level) {
 140       jcr->JobLevel = run->level;     /* override run level */
 141    }
 142    if (run->pool) {
 143       jcr->pool = run->pool;          /* override pool */
 144    }
 145    if (run->full_pool) {
 146       jcr->full_pool = run->full_pool; /* override full pool */
 147    }
 148    if (run->inc_pool) {
 149       jcr->inc_pool = run->inc_pool;  /* override inc pool */
 150    }
 151    if (run->dif_pool) {
 152       jcr->dif_pool = run->dif_pool;  /* override dif pool */
 153    }
 154    if (run->storage) {
 155       set_storage(jcr, run->storage); /* override storage */
 156    }
 157    if (run->msgs) {
 158       jcr->messages = run->msgs;      /* override messages */
 159    }
 160    if (run->Priority) {
 161       jcr->JobPriority = run->Priority;
 162    }
 163    if (run->spool_data_set) {
 164       jcr->spool_data = run->spool_data;
 165    }
 166    if (run->write_part_after_job_set) {
 167       jcr->write_part_after_job = run->write_part_after_job;
 168    }
 169    Dmsg0(200, "Leave wait_for_next_job()\n");
 170    return jcr;
 171 }
 172
 173
 174 /*
 175  * Shutdown the scheduler
 176  */
 177 void term_scheduler()
 178 {
 179    if (jobs_to_run) {
 180       job_item *je;
 181       /* Release all queued job entries to be run */
 182       foreach_dlist(je, jobs_to_run) {
 183          free(je);
 184       }
 185       delete jobs_to_run;
 186    }
 187 }
 188
 189 /*
 190  * Find all jobs to be run this hour and the next hour.
 191  */
 192 static void find_runs()
 193 {
 194    time_t now, next_hour, runtime;
 195    RUN *run;
 196    JOB *job;
 197    SCHED *sched;
 198    struct tm tm;
 199    int minute;
 200    int hour, mday, wday, month, wom, woy;
 201    /* Items corresponding to above at the next hour */
 202    int nh_hour, nh_mday, nh_wday, nh_month, nh_wom, nh_woy, nh_year;
 203
 204    Dmsg0(1200, "enter find_runs()\n");
 205
 206
 207    /* compute values for time now */
 208    now = time(NULL);
 209    localtime_r(&now, &tm);
 210    hour = tm.tm_hour;
 211    minute = tm.tm_min;
 212    mday = tm.tm_mday - 1;
 213    wday = tm.tm_wday;
 214    month = tm.tm_mon;
 215    wom = mday / 7;
 216    woy = tm_woy(now);                     /* get week of year */
 217
 218    /*
 219     * Compute values for next hour from now.
 220     * We do this to be sure we don't miss a job while
 221     * sleeping.
 222     */
 223    next_hour = now + 3600;
 224    localtime_r(&next_hour, &tm);
 225    nh_hour = tm.tm_hour;
 226    nh_mday = tm.tm_mday - 1;
 227    nh_wday = tm.tm_wday;
 228    nh_month = tm.tm_mon;
 229    nh_year  = tm.tm_year;
 230    nh_wom = nh_mday / 7;
 231    nh_woy = tm_woy(now);                     /* get week of year */
 232
 233    /* Loop through all jobs */
 234    LockRes();
 235    foreach_res(job, R_JOB) {
 236       sched = job->schedule;
 237       if (sched == NULL) {            /* scheduled? */
 238          continue;                    /* no, skip this job */
 239       }
 240       Dmsg1(1200, "Got job: %s\n", job->hdr.name);
 241       for (run=sched->run; run; run=run->next) {
 242          bool run_now, run_nh;
 243          /*
 244           * Find runs scheduled between now and the next hour.
 245           */
 246 #ifdef xxxx
 247          Dmsg0(000, "\n");
 248          Dmsg6(000, "run h=%d m=%d md=%d wd=%d wom=%d woy=%d\n",
 249             hour, month, mday, wday, wom, woy);
 250          Dmsg6(000, "bitset bsh=%d bsm=%d bsmd=%d bswd=%d bswom=%d bswoy=%d\n",
 251             bit_is_set(hour, run->hour),
 252             bit_is_set(month, run->month),
 253             bit_is_set(mday, run->mday),
 254             bit_is_set(wday, run->wday),
 255             bit_is_set(wom, run->wom),
 256             bit_is_set(woy, run->woy));
 257
 258          Dmsg6(000, "nh_run h=%d m=%d md=%d wd=%d wom=%d woy=%d\n",
 259             nh_hour, nh_month, nh_mday, nh_wday, nh_wom, nh_woy);
 260          Dmsg6(000, "nh_bitset bsh=%d bsm=%d bsmd=%d bswd=%d bswom=%d bswoy=%d\n",
 261             bit_is_set(nh_hour, run->hour),
 262             bit_is_set(nh_month, run->month),
 263             bit_is_set(nh_mday, run->mday),
 264             bit_is_set(nh_wday, run->wday),
 265             bit_is_set(nh_wom, run->wom),
 266             bit_is_set(nh_woy, run->woy));
 267 #endif
 268
 269          run_now = bit_is_set(hour, run->hour) &&
 270             bit_is_set(mday, run->mday) &&
 271             bit_is_set(wday, run->wday) &&
 272             bit_is_set(month, run->month) &&
 273             bit_is_set(wom, run->wom) &&
 274             bit_is_set(woy, run->woy);
 275
 276          run_nh = bit_is_set(nh_hour, run->hour) &&
 277             bit_is_set(nh_mday, run->mday) &&
 278             bit_is_set(nh_wday, run->wday) &&
 279             bit_is_set(nh_month, run->month) &&
 280             bit_is_set(nh_wom, run->wom) &&
 281             bit_is_set(nh_woy, run->woy);
 282
 283          Dmsg2(1200, "run_now=%d run_nh=%d\n", run_now, run_nh);
 284
 285          /* find time (time_t) job is to be run */
 286          localtime_r(&now, &tm);      /* reset tm structure */
 287          tm.tm_min = run->minute;     /* set run minute */
 288          tm.tm_sec = 0;               /* zero secs */
 289          if (run_now) {
 290             runtime = mktime(&tm);
 291             add_job(job, run, now, runtime);
 292          }
 293          /* If job is to be run in the next hour schedule it */
 294          if (run_nh) {
 295             /* Set correct values */
 296             tm.tm_hour = nh_hour;
 297             tm.tm_mday = nh_mday + 1; /* fixup because we biased for tests above */
 298             tm.tm_mon = nh_month;
 299             tm.tm_year = nh_year;
 300             runtime = mktime(&tm);
 301             add_job(job, run, now, runtime);
 302          }
 303       }
 304    }
 305    UnlockRes();
 306    Dmsg0(1200, "Leave find_runs()\n");
 307 }
 308
 309 static void add_job(JOB *job, RUN *run, time_t now, time_t runtime)
 310 {
 311    job_item *ji;
 312    bool inserted = false;
 313    /*
 314     * Don't run any job that ran less than a minute ago, but
 315     *  do run any job scheduled less than a minute ago.
 316     */
 317    if (((runtime - run->last_run) < 61) || ((runtime+59) < now)) {
 318 #ifdef SCHED_DEBUG
 319       char dt[50], dt1[50], dt2[50];
 320       bstrftime_nc(dt, sizeof(dt), runtime);
 321       bstrftime_nc(dt1, sizeof(dt1), run->last_run);
 322       bstrftime_nc(dt2, sizeof(dt2), now);
 323       Dmsg4(000, "Drop: Job=\"%s\" run=%s. last_run=%s. now=%s\n", job->hdr.name,
 324             dt, dt1, dt2);
 325       fflush(stdout);
 326 #endif
 327       return;
 328    }
 329    /* accept to run this job */
 330    job_item *je = (job_item *)malloc(sizeof(job_item));
 331    je->run = run;
 332    je->job = job;
 333    je->runtime = runtime;
 334    if (run->Priority) {
 335       je->Priority = run->Priority;
 336    } else {
 337       je->Priority = job->Priority;
 338    }
 339
 340    /* Add this job to the wait queue in runtime, priority sorted order */
 341    foreach_dlist(ji, jobs_to_run) {
 342       if (ji->runtime > je->runtime ||
 343           (ji->runtime == je->runtime && ji->Priority > je->Priority)) {
 344          jobs_to_run->insert_before(je, ji);
 345          dump_job(je, _("Inserted job"));
 346          inserted = true;
 347          break;
 348       }
 349    }
 350    /* If place not found in queue, append it */
 351    if (!inserted) {
 352       jobs_to_run->append(je);
 353       dump_job(je, _("Appended job"));
 354    }
 355 #ifdef SCHED_DEBUG
 356    foreach_dlist(ji, jobs_to_run) {
 357       dump_job(ji, _("Run queue"));
 358    }
 359    Dmsg0(000, "End run queue\n");
 360 #endif
 361 }
 362
 363 static void dump_job(job_item *ji, const char *msg)
 364 {
 365 #ifdef SCHED_DEBUG
 366    char dt[MAX_TIME_LENGTH];
 367    int save_debug = debug_level;
 368    debug_level = 200;
 369    if (debug_level < 200) {
 370       return;
 371    }
 372    bstrftime_nc(dt, sizeof(dt), ji->runtime);
 373    Dmsg4(200, "%s: Job=%s priority=%d run %s\n", msg, ji->job->hdr.name,
 374       ji->Priority, dt);
 375    fflush(stdout);
 376    debug_level = save_debug;
 377 #endif
 378 }