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

   1 /*
   2  *
   3  *   Bacula Director Job processing routines
   4  *
   5  *     Kern Sibbald, October MM
   6  *
   7  *    Version $Id$
   8  */
   9 /*
  10    Copyright (C) 2000-2003 Kern Sibbald and John Walker
  11
  12    This program is free software; you can redistribute it and/or
  13    modify it under the terms of the GNU General Public License as
  14    published by the Free Software Foundation; either version 2 of
  15    the License, or (at your option) any later version.
  16
  17    This program is distributed in the hope that it will be useful,
  18    but WITHOUT ANY WARRANTY; without even the implied warranty of
  19    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  20    General Public License for more details.
  21
  22    You should have received a copy of the GNU General Public
  23    License along with this program; if not, write to the Free
  24    Software Foundation, Inc., 59 Temple Place - Suite 330, Boston,
  25    MA 02111-1307, USA.
  26
  27  */
  28
  29 #include "bacula.h"
  30 #include "dird.h"
  31
  32 /* Forward referenced subroutines */
  33 static void *job_thread(void *arg);
  34 static char *edit_run_codes(JCR *jcr, char *omsg, char *imsg);
  35 static void release_resource_locks(JCR *jcr);
  36 static int acquire_resource_locks(JCR *jcr);
  37 #ifdef USE_SEMAPHORE
  38 static void backoff_resource_locks(JCR *jcr, int count);
  39 #endif
  40
  41 /* Exported subroutines */
  42 void run_job(JCR *jcr);
  43
  44
  45 /* Imported subroutines */
  46 extern void term_scheduler();
  47 extern void term_ua_server();
  48 extern int do_backup(JCR *jcr);
  49 extern int do_restore(JCR *jcr);
  50 extern int do_verify(JCR *jcr);
  51 extern void backup_cleanup(void);
  52
  53 #ifdef USE_SEMAPHORE
  54 static semlock_t job_lock;
  55 static pthread_mutex_t mutex;
  56 static pthread_cond_t  resource_wait;
  57 #else
  58 /* Queue of jobs to be run */
  59 workq_t job_wq;                   /* our job work queue */
  60 #endif
  61
  62 void init_job_server(int max_workers)
  63 {
  64    int stat;
  65 #ifdef USE_SEMAPHORE
  66    if ((stat = sem_init(&job_lock, max_workers)) != 0) {
  67       Emsg1(M_ABORT, 0, _("Could not init job lock: ERR=%s\n"), strerror(stat));
  68    }
  69    if ((stat = pthread_mutex_init(&mutex, NULL)) != 0) {
  70       Emsg1(M_ABORT, 0, _("Could not init resource mutex: ERR=%s\n"), strerror(stat));
  71    }
  72    if ((stat = pthread_cond_init(&resource_wait, NULL)) != 0) {
  73       Emsg1(M_ABORT, 0, _("Could not init resource wait: ERR=%s\n"), strerror(stat));
  74    }
  75
  76 #else
  77    if ((stat = workq_init(&job_wq, max_workers, job_thread)) != 0) {
  78       Emsg1(M_ABORT, 0, _("Could not init job work queue: ERR=%s\n"), strerror(stat));
  79    }
  80 #endif
  81    return;
  82 }
  83
  84 /*
  85  * Run a job -- typically called by the scheduler, but may also
  86  *              be called by the UA (Console program).
  87  *
  88  */
  89 void run_job(JCR *jcr)
  90 {
  91    int stat, errstat;
  92 #ifdef USE_SEMAPHORE
  93    pthread_t tid;
  94 #else
  95    workq_ele_t *work_item;
  96 #endif
  97
  98    sm_check(__FILE__, __LINE__, True);
  99    init_msg(jcr, jcr->messages);
 100    create_unique_job_name(jcr, jcr->job->hdr.name);
 101    jcr->jr.SchedTime = jcr->sched_time;
 102    jcr->jr.StartTime = jcr->start_time;
 103    jcr->jr.Type = jcr->JobType;
 104    jcr->jr.Level = jcr->JobLevel;
 105    jcr->jr.JobStatus = jcr->JobStatus;
 106    strcpy(jcr->jr.Name, jcr->job->hdr.name);
 107    strcpy(jcr->jr.Job, jcr->Job);
 108
 109    /* Initialize termination condition variable */
 110    if ((errstat = pthread_cond_init(&jcr->term_wait, NULL)) != 0) {
 111       Jmsg1(jcr, M_FATAL, 0, _("Unable to init job cond variable: ERR=%s\n"), strerror(errstat));
 112       set_jcr_job_status(jcr, JS_ErrorTerminated);
 113       free_jcr(jcr);
 114       return;
 115    }
 116
 117    /*
 118     * Open database
 119     */
 120    Dmsg0(50, "Open database\n");
 121    jcr->db=db_init_database(jcr, jcr->catalog->db_name, jcr->catalog->db_user,
 122                             jcr->catalog->db_password, jcr->catalog->db_address,
 123                             jcr->catalog->db_port, jcr->catalog->db_socket);
 124    if (!db_open_database(jcr, jcr->db)) {
 125       Jmsg(jcr, M_FATAL, 0, "%s", db_strerror(jcr->db));
 126       db_close_database(jcr, jcr->db);
 127       set_jcr_job_status(jcr, JS_ErrorTerminated);
 128       free_jcr(jcr);
 129       return;
 130    }
 131    Dmsg0(50, "DB opened\n");
 132
 133    /*
 134     * Create Job record
 135     */
 136    jcr->jr.JobStatus = jcr->JobStatus;
 137    if (!db_create_job_record(jcr, jcr->db, &jcr->jr)) {
 138       Jmsg(jcr, M_FATAL, 0, "%s", db_strerror(jcr->db));
 139       db_close_database(jcr, jcr->db);
 140       set_jcr_job_status(jcr, JS_ErrorTerminated);
 141       free_jcr(jcr);
 142       return;
 143    }
 144    jcr->JobId = jcr->jr.JobId;
 145    ASSERT(jcr->jr.JobId > 0);
 146
 147    Dmsg4(30, "Created job record JobId=%d Name=%s Type=%c Level=%c\n",
 148        jcr->JobId, jcr->Job, jcr->jr.Type, jcr->jr.Level);
 149    Dmsg0(200, "Add jrc to work queue\n");
 150
 151 #ifdef USE_SEMAPHORE
 152   if ((stat = pthread_create(&tid, NULL, job_thread, (void *)jcr)) != 0) {
 153       Emsg1(M_ABORT, 0, _("Unable to create job thread: ERR=%s\n"), strerror(stat));
 154    }
 155 #else
 156    /* Queue the job to be run */
 157    if ((stat = workq_add(&job_wq, (void *)jcr, &work_item, 0)) != 0) {
 158       Emsg1(M_ABORT, 0, _("Could not add job to work queue: ERR=%s\n"), strerror(stat));
 159    }
 160    jcr->work_item = work_item;
 161 #endif
 162    Dmsg0(200, "Done run_job()\n");
 163 }
 164
 165 /*
 166  * This is the engine called by workq_add() when we were pulled
 167  *  from the work queue.
 168  *  At this point, we are running in our own thread
 169  */
 170 static void *job_thread(void *arg)
 171 {
 172    time_t now;
 173    JCR *jcr = (JCR *)arg;
 174
 175    pthread_detach(pthread_self());
 176    time(&now);
 177    sm_check(__FILE__, __LINE__, True);
 178
 179    if (!acquire_resource_locks(jcr)) {
 180       set_jcr_job_status(jcr, JS_Cancelled);
 181    }
 182
 183    Dmsg0(200, "=====Start Job=========\n");
 184    jcr->start_time = now;             /* set the real start time */
 185    Dmsg2(200, "jcr->JobStatus=%d %c\n", jcr->JobStatus, (char)jcr->JobStatus);
 186    if (job_cancelled(jcr)) {
 187       update_job_end_record(jcr);
 188    } else if (jcr->job->MaxStartDelay != 0 && jcr->job->MaxStartDelay <
 189        (utime_t)(jcr->start_time - jcr->sched_time)) {
 190       Jmsg(jcr, M_FATAL, 0, _("Job cancelled because max start delay time exceeded.\n"));
 191       set_jcr_job_status(jcr, JS_Cancelled);
 192       update_job_end_record(jcr);
 193    } else {
 194
 195       /* Run Job */
 196       set_jcr_job_status(jcr, JS_Running);
 197
 198       if (jcr->job->RunBeforeJob) {
 199          POOLMEM *before = get_pool_memory(PM_FNAME);
 200          int status;
 201
 202          before = edit_run_codes(jcr, before, jcr->job->RunBeforeJob);
 203          status = run_program(before, 0, NULL);
 204          free_pool_memory(before);
 205       }
 206       switch (jcr->JobType) {
 207          case JT_BACKUP:
 208             do_backup(jcr);
 209             if (jcr->JobStatus == JS_Terminated) {
 210                do_autoprune(jcr);
 211             }
 212             break;
 213          case JT_VERIFY:
 214             do_verify(jcr);
 215             if (jcr->JobStatus == JS_Terminated) {
 216                do_autoprune(jcr);
 217             }
 218             break;
 219          case JT_RESTORE:
 220             do_restore(jcr);
 221             if (jcr->JobStatus == JS_Terminated) {
 222                do_autoprune(jcr);
 223             }
 224             break;
 225          case JT_ADMIN:
 226             /* No actual job */
 227             do_autoprune(jcr);
 228             set_jcr_job_status(jcr, JS_Terminated);
 229             break;
 230          default:
 231             Pmsg1(0, "Unimplemented job type: %d\n", jcr->JobType);
 232             break;
 233          }
 234       if (jcr->job->RunAfterJob) {
 235          POOLMEM *after = get_pool_memory(PM_FNAME);
 236          int status;
 237
 238          after = edit_run_codes(jcr, after, jcr->job->RunAfterJob);
 239          status = run_program(after, 0, NULL);
 240          free_pool_memory(after);
 241       }
 242    }
 243    release_resource_locks(jcr);
 244    Dmsg0(50, "Before free jcr\n");
 245    free_jcr(jcr);
 246    Dmsg0(50, "======== End Job ==========\n");
 247    sm_check(__FILE__, __LINE__, True);
 248    return NULL;
 249 }
 250
 251 static int acquire_resource_locks(JCR *jcr)
 252 {
 253 #ifdef USE_SEMAPHORE
 254    int stat;
 255
 256    if (jcr->store->sem.valid != SEMLOCK_VALID) {
 257       if ((stat = sem_init(&jcr->store->sem, jcr->store->MaxConcurrentJobs)) != 0) {
 258          Emsg1(M_ABORT, 0, _("Could not init Storage semaphore: ERR=%s\n"), strerror(stat));
 259       }
 260    }
 261    if (jcr->client->sem.valid != SEMLOCK_VALID) {
 262       if ((stat = sem_init(&jcr->client->sem, jcr->client->MaxConcurrentJobs)) != 0) {
 263          Emsg1(M_ABORT, 0, _("Could not init Client semaphore: ERR=%s\n"), strerror(stat));
 264       }
 265    }
 266    if (jcr->job->sem.valid != SEMLOCK_VALID) {
 267       if ((stat = sem_init(&jcr->job->sem, jcr->job->MaxConcurrentJobs)) != 0) {
 268          Emsg1(M_ABORT, 0, _("Could not init Job semaphore: ERR=%s\n"), strerror(stat));
 269       }
 270    }
 271
 272    for ( ;; ) {
 273       /* Acquire semaphore */
 274       set_jcr_job_status(jcr, JS_WaitJobRes);
 275       if ((stat = sem_lock(&jcr->job->sem)) != 0) {
 276          Emsg1(M_ABORT, 0, _("Could not acquire Job max jobs lock: ERR=%s\n"), strerror(stat));
 277       }
 278       set_jcr_job_status(jcr, JS_WaitClientRes);
 279       if ((stat = sem_trylock(&jcr->client->sem)) != 0) {
 280          if (stat == EBUSY) {
 281             backoff_resource_locks(jcr, 1);
 282             goto wait;
 283          } else {
 284             Emsg1(M_ABORT, 0, _("Could not acquire Client max jobs lock: ERR=%s\n"), strerror(stat));
 285          }
 286       }
 287       set_jcr_job_status(jcr, JS_WaitStoreRes);
 288       if ((stat = sem_trylock(&jcr->store->sem)) != 0) {
 289          if (stat == EBUSY) {
 290             backoff_resource_locks(jcr, 2);
 291             goto wait;
 292          } else {
 293             Emsg1(M_ABORT, 0, _("Could not acquire Storage max jobs lock: ERR=%s\n"), strerror(stat));
 294          }
 295       }
 296       set_jcr_job_status(jcr, JS_WaitMaxJobs);
 297       if ((stat = sem_trylock(&job_lock)) != 0) {
 298          if (stat == EBUSY) {
 299             backoff_resource_locks(jcr, 3);
 300             goto wait;
 301          } else {
 302             Emsg1(M_ABORT, 0, _("Could not acquire max jobs lock: ERR=%s\n"), strerror(stat));
 303          }
 304       }
 305       break;
 306
 307 wait:
 308       P(mutex);
 309       /* Wait for some resource to be released */
 310       pthread_cond_wait(&resource_wait, &mutex);
 311       V(mutex);
 312       /* Try again */
 313    }
 314 #endif
 315    return 1;
 316 }
 317
 318 #ifdef USE_SEMAPHORE
 319 static void backoff_resource_locks(JCR *jcr, int count)
 320 {
 321    switch (count) {
 322    case 3:
 323       sem_unlock(&jcr->store->sem);
 324    case 2:
 325       sem_unlock(&jcr->client->sem);
 326    case 1:
 327       sem_unlock(&jcr->job->sem);
 328       break;
 329    }
 330 }
 331 #endif
 332
 333 static void release_resource_locks(JCR *jcr)
 334 {
 335 #ifdef USE_SEMAPHORE
 336    P(mutex);
 337    sem_unlock(&jcr->store->sem);
 338    sem_unlock(&jcr->client->sem);
 339    sem_unlock(&jcr->job->sem);
 340    sem_unlock(&job_lock);
 341    pthread_cond_signal(&resource_wait);
 342    V(mutex);
 343 #endif
 344 }
 345
 346 /*
 347  * Get or create a Client record for this Job
 348  */
 349 int get_or_create_client_record(JCR *jcr)
 350 {
 351    CLIENT_DBR cr;
 352
 353    memset(&cr, 0, sizeof(cr));
 354    strcpy(cr.Name, jcr->client->hdr.name);
 355    cr.AutoPrune = jcr->client->AutoPrune;
 356    cr.FileRetention = jcr->client->FileRetention;
 357    cr.JobRetention = jcr->client->JobRetention;
 358    if (jcr->client_name) {
 359       free_pool_memory(jcr->client_name);
 360    }
 361    jcr->client_name = get_memory(strlen(jcr->client->hdr.name) + 1);
 362    strcpy(jcr->client_name, jcr->client->hdr.name);
 363    if (!db_create_client_record(jcr, jcr->db, &cr)) {
 364       Jmsg(jcr, M_FATAL, 0, _("Could not create Client record. %s"),
 365          db_strerror(jcr->db));
 366       return 0;
 367    }
 368    jcr->jr.ClientId = cr.ClientId;
 369    if (cr.Uname[0]) {
 370       if (jcr->client_uname) {
 371          free_pool_memory(jcr->client_uname);
 372       }
 373       jcr->client_uname = get_memory(strlen(cr.Uname) + 1);
 374       strcpy(jcr->client_uname, cr.Uname);
 375    }
 376    Dmsg2(100, "Created Client %s record %d\n", jcr->client->hdr.name,
 377       jcr->jr.ClientId);
 378    return 1;
 379 }
 380
 381
 382 /*
 383  * Write status and such in DB
 384  */
 385 void update_job_end_record(JCR *jcr)
 386 {
 387    if (jcr->jr.EndTime == 0) {
 388       jcr->jr.EndTime = time(NULL);
 389    }
 390    jcr->end_time = jcr->jr.EndTime;
 391    jcr->jr.JobId = jcr->JobId;
 392    jcr->jr.JobStatus = jcr->JobStatus;
 393    jcr->jr.JobFiles = jcr->JobFiles;
 394    jcr->jr.JobBytes = jcr->JobBytes;
 395    jcr->jr.VolSessionId = jcr->VolSessionId;
 396    jcr->jr.VolSessionTime = jcr->VolSessionTime;
 397    if (!db_update_job_end_record(jcr, jcr->db, &jcr->jr)) {
 398       Jmsg(jcr, M_WARNING, 0, _("Error updating job record. %s"),
 399          db_strerror(jcr->db));
 400    }
 401 }
 402
 403 /*
 404  * Takes base_name and appends (unique) current
 405  *   date and time to form unique job name.
 406  *
 407  *  Returns: unique job name in jcr->Job
 408  *    date/time in jcr->start_time
 409  */
 410 void create_unique_job_name(JCR *jcr, char *base_name)
 411 {
 412    /* Job start mutex */
 413    static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
 414    static time_t last_start_time = 0;
 415    time_t now;
 416    struct tm tm;
 417    char dt[MAX_TIME_LENGTH];
 418    char name[MAX_NAME_LENGTH];
 419    char *p;
 420
 421    /* Guarantee unique start time -- maximum one per second, and
 422     * thus unique Job Name
 423     */
 424    P(mutex);                          /* lock creation of jobs */
 425    now = time(NULL);
 426    while (now == last_start_time) {
 427       sleep(1);
 428       now = time(NULL);
 429    }
 430    last_start_time = now;
 431    V(mutex);                          /* allow creation of jobs */
 432    jcr->start_time = now;
 433    /* Form Unique JobName */
 434    localtime_r(&now, &tm);
 435    /* Use only characters that are permitted in Windows filenames */
 436    strftime(dt, sizeof(dt), "%Y-%m-%d_%H.%M.%S", &tm);
 437    bstrncpy(name, base_name, sizeof(name));
 438    name[sizeof(name)-22] = 0;          /* truncate if too long */
 439    sprintf(jcr->Job, "%s.%s", name, dt); /* add date & time */
 440    /* Convert spaces into underscores */
 441    for (p=jcr->Job; *p; p++) {
 442       if (*p == ' ') {
 443          *p = '_';
 444       }
 445    }
 446 }
 447
 448 /*
 449  * Free the Job Control Record if no one is still using it.
 450  *  Called from main free_jcr() routine in src/lib/jcr.c so
 451  *  that we can do our Director specific cleanup of the jcr.
 452  */
 453 void dird_free_jcr(JCR *jcr)
 454 {
 455    Dmsg0(200, "Start dird free_jcr\n");
 456
 457    if (jcr->file_bsock) {
 458       Dmsg0(200, "Close File bsock\n");
 459       bnet_close(jcr->file_bsock);
 460    }
 461    if (jcr->store_bsock) {
 462       Dmsg0(200, "Close Store bsock\n");
 463       bnet_close(jcr->store_bsock);
 464    }
 465    if (jcr->fname) {
 466       Dmsg0(200, "Free JCR fname\n");
 467       free_pool_memory(jcr->fname);
 468    }
 469    if (jcr->stime) {
 470       Dmsg0(200, "Free JCR stime\n");
 471       free_pool_memory(jcr->stime);
 472    }
 473    if (jcr->db) {
 474       Dmsg0(200, "Close DB\n");
 475       db_close_database(jcr, jcr->db);
 476    }
 477    if (jcr->RestoreWhere) {
 478       free(jcr->RestoreWhere);
 479    }
 480    if (jcr->RestoreBootstrap) {
 481       free(jcr->RestoreBootstrap);
 482    }
 483    Dmsg0(200, "End dird free_jcr\n");
 484 }
 485
 486 /*
 487  * Set some defaults in the JCR necessary to
 488  * run. These items are pulled from the job
 489  * definition as defaults, but can be overridden
 490  * later either by the Run record in the Schedule resource,
 491  * or by the Console program.
 492  */
 493 void set_jcr_defaults(JCR *jcr, JOB *job)
 494 {
 495    jcr->job = job;
 496    jcr->JobType = job->JobType;
 497    jcr->JobLevel = job->level;
 498    jcr->store = job->storage;
 499    jcr->client = job->client;
 500    if (jcr->client_name) {
 501       free_pool_memory(jcr->client_name);
 502    }
 503    jcr->client_name = get_memory(strlen(jcr->client->hdr.name) + 1);
 504    strcpy(jcr->client_name, jcr->client->hdr.name);
 505    jcr->pool = job->pool;
 506    jcr->catalog = job->client->catalog;
 507    jcr->fileset = job->fileset;
 508    jcr->messages = job->messages;
 509    if (jcr->RestoreBootstrap) {
 510       free(jcr->RestoreBootstrap);
 511    }
 512    /* This can be overridden by Console program */
 513    if (job->RestoreBootstrap) {
 514       jcr->RestoreBootstrap = bstrdup(job->RestoreBootstrap);
 515    }
 516    /* If no default level given, set one */
 517    if (jcr->JobLevel == 0) {
 518       switch (jcr->JobType) {
 519       case JT_VERIFY:
 520          jcr->JobLevel = L_VERIFY_CATALOG;
 521          break;
 522       case JT_BACKUP:
 523          jcr->JobLevel = L_INCREMENTAL;
 524          break;
 525       default:
 526          break;
 527       }
 528    }
 529 }
 530
 531 /*
 532  * Edit codes into Run command
 533  *  %% = %
 534  *  %c = Client's name
 535  *  %d = Director's name
 536  *  %i = JobId
 537  *  %e = Job Exit
 538  *  %j = Job
 539  *  %l = Job Level
 540  *  %n = Job name
 541  *  %t = Job type
 542  *
 543  *  omsg = edited output message
 544  *  imsg = input string containing edit codes (%x)
 545  *
 546  */
 547 static char *edit_run_codes(JCR *jcr, char *omsg, char *imsg)
 548 {
 549    char *p;
 550    const char *str;
 551    char add[20];
 552
 553    *omsg = 0;
 554    Dmsg1(200, "edit_run_codes: %s\n", imsg);
 555    for (p=imsg; *p; p++) {
 556       if (*p == '%') {
 557          switch (*++p) {
 558          case '%':
 559             str = "%";
 560             break;
 561          case 'c':
 562             str = jcr->client_name;
 563             if (!str) {
 564                str = "";
 565             }
 566             break;
 567          case 'd':
 568             str = my_name;
 569             break;
 570          case 'e':
 571             str = job_status_to_str(jcr->JobStatus);
 572             break;
 573          case 'i':
 574             sprintf(add, "%d", jcr->JobId);
 575             str = add;
 576             break;
 577          case 'j':                    /* Job */
 578             str = jcr->Job;
 579             break;
 580          case 'l':
 581             str = job_level_to_str(jcr->JobLevel);
 582             break;
 583          case 'n':
 584             str = jcr->job->hdr.name;
 585             break;
 586          case 't':
 587             str = job_type_to_str(jcr->JobType);
 588             break;
 589          default:
 590             add[0] = '%';
 591             add[1] = *p;
 592             add[2] = 0;
 593             str = add;
 594             break;
 595          }
 596       } else {
 597          add[0] = *p;
 598          add[1] = 0;
 599          str = add;
 600       }
 601       Dmsg1(200, "add_str %s\n", str);
 602       pm_strcat(&omsg, (char *)str);
 603       Dmsg1(200, "omsg=%s\n", omsg);
 604    }
 605    return omsg;
 606 }