git.sur5r.net Git - bacula/bacula/blob - bacula/src/lib/jcr.c

   1 /*
   2  * Manipulation routines for Job Control Records and
   3  *  handling of last_jobs_list.
   4  *
   5  *  Kern E. Sibbald, December 2000
   6  *
   7  *  Version $Id$
   8  *
   9  *  These routines are thread safe.
  10  *
  11  *  The job list routines were re-written in May 2005 to
  12  *  eliminate the global lock while traversing the list, and
  13  *  to use the dlist subroutines.  The locking is now done
  14  *  on the list each time the list is modified or traversed.
  15  *  That is it is "micro-locked" rather than globally locked.
  16  *  The result is that there is one lock/unlock for each entry
  17  *  in the list while traversing it rather than a single lock
  18  *  at the beginning of a traversal and one at the end.  This
  19  *  incurs slightly more overhead, but effectively eliminates
  20  *  the possibilty of race conditions.  In addition, with the
  21  *  exception of the global locking of the list during the
  22  *  re-reading of the config file, no recursion is needed.
  23  *
  24  */
  25 /*
  26    Copyright (C) 2000-2005 Kern Sibbald
  27
  28    This program is free software; you can redistribute it and/or
  29    modify it under the terms of the GNU General Public License
  30    version 2 as amended with additional clauses defined in the
  31    file LICENSE in the main source directory.
  32
  33    This program is distributed in the hope that it will be useful,
  34    but WITHOUT ANY WARRANTY; without even the implied warranty of
  35    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  36    the file LICENSE for additional details.
  37
  38  */
  39
  40 #include "bacula.h"
  41 #include "jcr.h"
  42
  43 /* External variables we reference */
  44 extern time_t watchdog_time;
  45
  46 /* Forward referenced functions */
  47 extern "C" void timeout_handler(int sig);
  48 static void jcr_timeout_check(watchdog_t *self);
  49 #ifdef TRACE_JCR_CHAIN
  50 static void b_lock_jcr_chain(const char *filen, int line);
  51 static void b_unlock_jcr_chain(const char *filen, int line);
  52 #define lock_jcr_chain() b_lock_jcr_chain(__FILE__, __LINE__);
  53 #define unlock_jcr_chain() b_unlock_jcr_chain(__FILE__, __LINE__);
  54 #else
  55 static void lock_jcr_chain();
  56 static void unlock_jcr_chain();
  57 #endif
  58
  59
  60 int num_jobs_run;
  61 dlist *last_jobs = NULL;
  62 const int max_last_jobs = 10;
  63
  64 static dlist *jcrs = NULL;            /* JCR chain */
  65 static pthread_mutex_t jcr_lock = PTHREAD_MUTEX_INITIALIZER;
  66
  67 static pthread_mutex_t job_start_mutex = PTHREAD_MUTEX_INITIALIZER;
  68
  69 void lock_jobs()
  70 {
  71    P(job_start_mutex);
  72 }
  73
  74 void unlock_jobs()
  75 {
  76    V(job_start_mutex);
  77 }
  78
  79 void init_last_jobs_list()
  80 {
  81    JCR *jcr = NULL;
  82    struct s_last_job *job_entry = NULL;
  83    if (!last_jobs) {
  84       last_jobs = New(dlist(job_entry, &job_entry->link));
  85    }
  86    if (!jcrs) {
  87       jcrs = New(dlist(jcr, &jcr->link));
  88    }
  89 }
  90
  91 void term_last_jobs_list()
  92 {
  93    if (last_jobs) {
  94       while (!last_jobs->empty()) {
  95          void *je = last_jobs->first();
  96          last_jobs->remove(je);
  97          free(je);
  98       }
  99       delete last_jobs;
 100       last_jobs = NULL;
 101    }
 102    if (jcrs) {
 103       delete jcrs;
 104       jcrs = NULL;
 105    }
 106 }
 107
 108 void read_last_jobs_list(int fd, uint64_t addr)
 109 {
 110    struct s_last_job *je, job;
 111    uint32_t num;
 112
 113    Dmsg1(100, "read_last_jobs seek to %d\n", (int)addr);
 114    if (addr == 0 || lseek(fd, (off_t)addr, SEEK_SET) < 0) {
 115       return;
 116    }
 117    if (read(fd, &num, sizeof(num)) != sizeof(num)) {
 118       return;
 119    }
 120    Dmsg1(100, "Read num_items=%d\n", num);
 121    if (num > 4 * max_last_jobs) {  /* sanity check */
 122       return;
 123    }
 124    for ( ; num; num--) {
 125       if (read(fd, &job, sizeof(job)) != sizeof(job)) {
 126          Dmsg1(000, "Read job entry. ERR=%s\n", strerror(errno));
 127          return;
 128       }
 129       if (job.JobId > 0) {
 130          je = (struct s_last_job *)malloc(sizeof(struct s_last_job));
 131          memcpy((char *)je, (char *)&job, sizeof(job));
 132          if (!last_jobs) {
 133             init_last_jobs_list();
 134          }
 135          last_jobs->append(je);
 136          if (last_jobs->size() > max_last_jobs) {
 137             je = (struct s_last_job *)last_jobs->first();
 138             last_jobs->remove(je);
 139             free(je);
 140          }
 141       }
 142    }
 143 }
 144
 145 uint64_t write_last_jobs_list(int fd, uint64_t addr)
 146 {
 147    struct s_last_job *je;
 148    uint32_t num;
 149
 150    Dmsg1(100, "write_last_jobs seek to %d\n", (int)addr);
 151    if (lseek(fd, (off_t)addr, SEEK_SET) < 0) {
 152       return 0;
 153    }
 154    if (last_jobs) {
 155       /* First record is number of entires */
 156       num = last_jobs->size();
 157       if (write(fd, &num, sizeof(num)) != sizeof(num)) {
 158          Dmsg1(000, "Error writing num_items: ERR=%s\n", strerror(errno));
 159          return 0;
 160       }
 161       foreach_dlist(je, last_jobs) {
 162          if (write(fd, je, sizeof(struct s_last_job)) != sizeof(struct s_last_job)) {
 163             Dmsg1(000, "Error writing job: ERR=%s\n", strerror(errno));
 164             return 0;
 165          }
 166       }
 167    }
 168    /* Return current address */
 169    ssize_t stat = lseek(fd, 0, SEEK_CUR);
 170    if (stat < 0) {
 171       stat = 0;
 172    }
 173    return stat;
 174
 175 }
 176
 177 void lock_last_jobs_list()
 178 {
 179    /* Use jcr chain mutex */
 180    lock_jcr_chain();
 181 }
 182
 183 void unlock_last_jobs_list()
 184 {
 185    /* Use jcr chain mutex */
 186    unlock_jcr_chain();
 187 }
 188
 189 /*
 190  * Push a subroutine address into the job end callback stack
 191  */
 192 void job_end_push(JCR *jcr, void job_end_cb(JCR *jcr,void *), void *ctx)
 193 {
 194    jcr->job_end_push.append((void *)job_end_cb);
 195    jcr->job_end_push.append(ctx);
 196 }
 197
 198 /* Pop each job_end subroutine and call it */
 199 static void job_end_pop(JCR *jcr)
 200 {
 201    void (*job_end_cb)(JCR *jcr, void *ctx);
 202    void *ctx;
 203    for (int i=jcr->job_end_push.size()-1; i > 0; ) {
 204       ctx = jcr->job_end_push.get(i--);
 205       job_end_cb = (void (*)(JCR *,void *))jcr->job_end_push.get(i--);
 206       job_end_cb(jcr, ctx);
 207    }
 208 }
 209
 210 /*
 211  * Create a Job Control Record and link it into JCR chain
 212  * Returns newly allocated JCR
 213  * Note, since each daemon has a different JCR, he passes
 214  *  us the size.
 215  */
 216 JCR *new_jcr(int size, JCR_free_HANDLER *daemon_free_jcr)
 217 {
 218    JCR *jcr;
 219    MQUEUE_ITEM *item = NULL;
 220    struct sigaction sigtimer;
 221
 222    Dmsg0(3400, "Enter new_jcr\n");
 223    jcr = (JCR *)malloc(size);
 224    memset(jcr, 0, size);
 225    jcr->my_thread_id = pthread_self();
 226    jcr->msg_queue = New(dlist(item, &item->link));
 227    jcr->job_end_push.init(1, false);
 228    jcr->sched_time = time(NULL);
 229    jcr->daemon_free_jcr = daemon_free_jcr;    /* plug daemon free routine */
 230    jcr->use_count = 1;
 231    pthread_mutex_init(&(jcr->mutex), NULL);
 232    jcr->JobStatus = JS_Created;       /* ready to run */
 233    jcr->VolumeName = get_pool_memory(PM_FNAME);
 234    jcr->VolumeName[0] = 0;
 235    jcr->errmsg = get_pool_memory(PM_MESSAGE);
 236    jcr->errmsg[0] = 0;
 237    /* Setup some dummy values */
 238    bstrncpy(jcr->Job, "*System*", sizeof(jcr->Job));
 239    jcr->JobId = 0;
 240    jcr->JobType = JT_SYSTEM;          /* internal job until defined */
 241    jcr->JobLevel = L_NONE;
 242    jcr->JobStatus = JS_Created;
 243
 244    sigtimer.sa_flags = 0;
 245    sigtimer.sa_handler = timeout_handler;
 246    sigfillset(&sigtimer.sa_mask);
 247    sigaction(TIMEOUT_SIGNAL, &sigtimer, NULL);
 248
 249    /*
 250     * Locking jobs is a global lock that is needed
 251     * so that the Director can stop new jobs from being
 252     * added to the jcr chain while it processes a new
 253     * conf file and does the job_end_push().
 254     */
 255    lock_jobs();
 256    lock_jcr_chain();
 257    if (!jcrs) {
 258       jcrs = New(dlist(jcr, &jcr->link));
 259    }
 260    jcrs->append(jcr);
 261    unlock_jcr_chain();
 262    unlock_jobs();
 263
 264    return jcr;
 265 }
 266
 267
 268 /*
 269  * Remove a JCR from the chain
 270  * NOTE! The chain must be locked prior to calling
 271  *       this routine.
 272  */
 273 static void remove_jcr(JCR *jcr)
 274 {
 275    Dmsg0(3400, "Enter remove_jcr\n");
 276    if (!jcr) {
 277       Emsg0(M_ABORT, 0, _("NULL jcr.\n"));
 278    }
 279    jcrs->remove(jcr);
 280    Dmsg0(3400, "Leave remove_jcr\n");
 281 }
 282
 283 /*
 284  * Free stuff common to all JCRs.  N.B. Be careful to include only
 285  *  generic stuff in the common part of the jcr.
 286  */
 287 static void free_common_jcr(JCR *jcr)
 288 {
 289    struct s_last_job *je, last_job;
 290
 291    /* Keep some statistics */
 292    switch (jcr->JobType) {
 293    case JT_BACKUP:
 294    case JT_VERIFY:
 295    case JT_RESTORE:
 296    case JT_ADMIN:
 297       num_jobs_run++;
 298       last_job.Errors = jcr->Errors;
 299       last_job.JobType = jcr->JobType;
 300       last_job.JobId = jcr->JobId;
 301       last_job.VolSessionId = jcr->VolSessionId;
 302       last_job.VolSessionTime = jcr->VolSessionTime;
 303       bstrncpy(last_job.Job, jcr->Job, sizeof(last_job.Job));
 304       last_job.JobFiles = jcr->JobFiles;
 305       last_job.JobBytes = jcr->JobBytes;
 306       last_job.JobStatus = jcr->JobStatus;
 307       last_job.JobLevel = jcr->JobLevel;
 308       last_job.start_time = jcr->start_time;
 309       last_job.end_time = time(NULL);
 310       /* Keep list of last jobs, but not Console where JobId==0 */
 311       if (last_job.JobId > 0) {
 312          je = (struct s_last_job *)malloc(sizeof(struct s_last_job));
 313          memcpy((char *)je, (char *)&last_job, sizeof(last_job));
 314          if (!last_jobs) {
 315             init_last_jobs_list();
 316          }
 317          last_jobs->append(je);
 318          if (last_jobs->size() > max_last_jobs) {
 319             je = (struct s_last_job *)last_jobs->first();
 320             last_jobs->remove(je);
 321             free(je);
 322          }
 323       }
 324       break;
 325    default:
 326       break;
 327    }
 328    pthread_mutex_destroy(&jcr->mutex);
 329
 330    delete jcr->msg_queue;
 331    close_msg(jcr);                    /* close messages for this job */
 332
 333    /* do this after closing messages */
 334    if (jcr->client_name) {
 335       free_pool_memory(jcr->client_name);
 336       jcr->client_name = NULL;
 337    }
 338
 339    if (jcr->attr) {
 340       free_pool_memory(jcr->attr);
 341       jcr->attr = NULL;
 342    }
 343
 344    if (jcr->sd_auth_key) {
 345       free(jcr->sd_auth_key);
 346       jcr->sd_auth_key = NULL;
 347    }
 348    if (jcr->VolumeName) {
 349       free_pool_memory(jcr->VolumeName);
 350       jcr->VolumeName = NULL;
 351    }
 352
 353    if (jcr->dir_bsock) {
 354       bnet_close(jcr->dir_bsock);
 355       jcr->dir_bsock = NULL;
 356    }
 357    if (jcr->errmsg) {
 358       free_pool_memory(jcr->errmsg);
 359       jcr->errmsg = NULL;
 360    }
 361    if (jcr->where) {
 362       free(jcr->where);
 363       jcr->where = NULL;
 364    }
 365    if (jcr->cached_path) {
 366       free_pool_memory(jcr->cached_path);
 367       jcr->cached_path = NULL;
 368       jcr->cached_pnl = 0;
 369    }
 370    free_getuser_cache();
 371    free_getgroup_cache();
 372    free(jcr);
 373 }
 374
 375 /*
 376  * Global routine to free a jcr
 377  */
 378 #ifdef DEBUG
 379 void b_free_jcr(const char *file, int line, JCR *jcr)
 380 {
 381    Dmsg3(3400, "Enter free_jcr 0x%x from %s:%d\n", jcr, file, line);
 382
 383 #else
 384
 385 void free_jcr(JCR *jcr)
 386 {
 387
 388    Dmsg1(3400, "Enter free_jcr 0x%x\n", jcr);
 389
 390 #endif
 391
 392    dequeue_messages(jcr);
 393    lock_jcr_chain();
 394    jcr->dec_use_count();              /* decrement use count */
 395    if (jcr->use_count < 0) {
 396       Emsg2(M_ERROR, 0, _("JCR use_count=%d JobId=%d\n"),
 397          jcr->use_count, jcr->JobId);
 398    }
 399    Dmsg3(3400, "Dec free_jcr 0x%x use_count=%d jobid=%d\n", jcr, jcr->use_count, jcr->JobId);
 400    if (jcr->use_count > 0) {          /* if in use */
 401       unlock_jcr_chain();
 402       Dmsg2(3400, "free_jcr 0x%x use_count=%d\n", jcr, jcr->use_count);
 403       return;
 404    }
 405
 406    remove_jcr(jcr);                   /* remove Jcr from chain */
 407    unlock_jcr_chain();
 408
 409    job_end_pop(jcr);                  /* pop and call hooked routines */
 410
 411    Dmsg1(3400, "End job=%d\n", jcr->JobId);
 412    if (jcr->daemon_free_jcr) {
 413       jcr->daemon_free_jcr(jcr);      /* call daemon free routine */
 414    }
 415    free_common_jcr(jcr);
 416    close_msg(NULL);                   /* flush any daemon messages */
 417    Dmsg0(3400, "Exit free_jcr\n");
 418 }
 419
 420
 421 /*
 422  * Given a JobId, find the JCR
 423  *   Returns: jcr on success
 424  *            NULL on failure
 425  */
 426 JCR *get_jcr_by_id(uint32_t JobId)
 427 {
 428    JCR *jcr;
 429
 430    lock_jcr_chain();                    /* lock chain */
 431    foreach_dlist(jcr, jcrs) {
 432       if (jcr->JobId == JobId) {
 433          jcr->inc_use_count();
 434          Dmsg2(3400, "Inc get_jcr 0x%x use_count=%d\n", jcr, jcr->use_count);
 435          break;
 436       }
 437    }
 438    unlock_jcr_chain();
 439    return jcr;
 440 }
 441
 442 /*
 443  * Given a SessionId and SessionTime, find the JCR
 444  *   Returns: jcr on success
 445  *            NULL on failure
 446  */
 447 JCR *get_jcr_by_session(uint32_t SessionId, uint32_t SessionTime)
 448 {
 449    JCR *jcr;
 450
 451    lock_jcr_chain();
 452    foreach_dlist(jcr, jcrs) {
 453       if (jcr->VolSessionId == SessionId &&
 454           jcr->VolSessionTime == SessionTime) {
 455          jcr->inc_use_count();
 456          Dmsg2(3400, "Inc get_jcr 0x%x use_count=%d\n", jcr, jcr->use_count);
 457          break;
 458       }
 459    }
 460    unlock_jcr_chain();
 461    return jcr;
 462 }
 463
 464
 465 /*
 466  * Given a Job, find the JCR
 467  *  compares on the number of characters in Job
 468  *  thus allowing partial matches.
 469  *   Returns: jcr on success
 470  *            NULL on failure
 471  */
 472 JCR *get_jcr_by_partial_name(char *Job)
 473 {
 474    JCR *jcr;
 475    int len;
 476
 477    if (!Job) {
 478       return NULL;
 479    }
 480    lock_jcr_chain();
 481    len = strlen(Job);
 482    foreach_dlist(jcr, jcrs) {
 483       if (strncmp(Job, jcr->Job, len) == 0) {
 484          jcr->inc_use_count();
 485          Dmsg2(3400, "Inc get_jcr 0x%x use_count=%d\n", jcr, jcr->use_count);
 486          break;
 487       }
 488    }
 489    unlock_jcr_chain();
 490    return jcr;
 491 }
 492
 493
 494 /*
 495  * Given a Job, find the JCR
 496  *  requires an exact match of names.
 497  *   Returns: jcr on success
 498  *            NULL on failure
 499  */
 500 JCR *get_jcr_by_full_name(char *Job)
 501 {
 502    JCR *jcr;
 503
 504    if (!Job) {
 505       return NULL;
 506    }
 507    lock_jcr_chain();
 508    foreach_dlist(jcr, jcrs) {
 509       if (strcmp(jcr->Job, Job) == 0) {
 510          jcr->inc_use_count();
 511          Dmsg2(3400, "Inc get_jcr 0x%x use_count=%d\n", jcr, jcr->use_count);
 512          break;
 513       }
 514    }
 515    unlock_jcr_chain();
 516    return jcr;
 517 }
 518
 519 void set_jcr_job_status(JCR *jcr, int JobStatus)
 520 {
 521    /*
 522     * For a set of errors, ... keep the current status
 523     *   so it isn't lost. For all others, set it.
 524     */
 525    switch (jcr->JobStatus) {
 526    case JS_ErrorTerminated:
 527    case JS_Error:
 528    case JS_FatalError:
 529    case JS_Differences:
 530    case JS_Canceled:
 531       break;
 532    default:
 533       jcr->JobStatus = JobStatus;
 534    }
 535 }
 536
 537 #ifdef TRACE_JCR_CHAIN
 538 static int lock_count = 0;
 539 #endif
 540
 541 /*
 542  * Lock the chain
 543  */
 544 #ifdef TRACE_JCR_CHAIN
 545 static void b_lock_jcr_chain(const char *fname, int line)
 546 #else
 547 static void lock_jcr_chain()
 548 #endif
 549 {
 550 #ifdef TRACE_JCR_CHAIN
 551    Dmsg3(3400, "Lock jcr chain %d from %s:%d\n", ++lock_count,
 552       fname, line);
 553 #endif
 554    P(jcr_lock);
 555 }
 556
 557 /*
 558  * Unlock the chain
 559  */
 560 #ifdef TRACE_JCR_CHAIN
 561 static void b_unlock_jcr_chain(const char *fname, int line)
 562 #else
 563 static void unlock_jcr_chain()
 564 #endif
 565 {
 566 #ifdef TRACE_JCR_CHAIN
 567    Dmsg3(3400, "Unlock jcr chain %d from %s:%d\n", lock_count--,
 568       fname, line);
 569 #endif
 570    V(jcr_lock);
 571 }
 572
 573
 574 JCR *get_next_jcr(JCR *prev_jcr)
 575 {
 576    JCR *jcr;
 577
 578    lock_jcr_chain();
 579    jcr = (JCR *)jcrs->next(prev_jcr);
 580    if (jcr) {
 581       jcr->inc_use_count();
 582       Dmsg2(3400, "Inc get_next_jcr 0x%x use_count=%d\n", jcr, jcr->use_count);
 583    }
 584    unlock_jcr_chain();
 585    return jcr;
 586 }
 587
 588 bool init_jcr_subsystem(void)
 589 {
 590    watchdog_t *wd = new_watchdog();
 591
 592    wd->one_shot = false;
 593    wd->interval = 30;   /* FIXME: should be configurable somewhere, even
 594                          if only with a #define */
 595    wd->callback = jcr_timeout_check;
 596
 597    register_watchdog(wd);
 598
 599    return true;
 600 }
 601
 602 static void jcr_timeout_check(watchdog_t *self)
 603 {
 604    JCR *jcr;
 605    BSOCK *fd;
 606    time_t timer_start;
 607
 608    Dmsg0(3400, "Start JCR timeout checks\n");
 609
 610    /* Walk through all JCRs checking if any one is
 611     * blocked for more than specified max time.
 612     */
 613    foreach_jcr(jcr) {
 614       if (jcr->JobId == 0) {
 615          free_jcr(jcr);
 616          continue;
 617       }
 618       fd = jcr->store_bsock;
 619       if (fd) {
 620          timer_start = fd->timer_start;
 621          if (timer_start && (watchdog_time - timer_start) > fd->timeout) {
 622             fd->timer_start = 0;      /* turn off timer */
 623             fd->timed_out = TRUE;
 624             Jmsg(jcr, M_ERROR, 0, _(
 625 "Watchdog sending kill after %d secs to thread stalled reading Storage daemon.\n"),
 626                  watchdog_time - timer_start);
 627             pthread_kill(jcr->my_thread_id, TIMEOUT_SIGNAL);
 628          }
 629       }
 630       fd = jcr->file_bsock;
 631       if (fd) {
 632          timer_start = fd->timer_start;
 633          if (timer_start && (watchdog_time - timer_start) > fd->timeout) {
 634             fd->timer_start = 0;      /* turn off timer */
 635             fd->timed_out = TRUE;
 636             Jmsg(jcr, M_ERROR, 0, _(
 637 "Watchdog sending kill after %d secs to thread stalled reading File daemon.\n"),
 638                  watchdog_time - timer_start);
 639             pthread_kill(jcr->my_thread_id, TIMEOUT_SIGNAL);
 640          }
 641       }
 642       fd = jcr->dir_bsock;
 643       if (fd) {
 644          timer_start = fd->timer_start;
 645          if (timer_start && (watchdog_time - timer_start) > fd->timeout) {
 646             fd->timer_start = 0;      /* turn off timer */
 647             fd->timed_out = TRUE;
 648             Jmsg(jcr, M_ERROR, 0, _(
 649 "Watchdog sending kill after %d secs to thread stalled reading Director.\n"),
 650                  watchdog_time - timer_start);
 651             pthread_kill(jcr->my_thread_id, TIMEOUT_SIGNAL);
 652          }
 653       }
 654       free_jcr(jcr);
 655    }
 656
 657    Dmsg0(3400, "Finished JCR timeout checks\n");
 658 }
 659
 660 /*
 661  * Timeout signal comes here
 662  */
 663 extern "C" void timeout_handler(int sig)
 664 {
 665    return;                            /* thus interrupting the function */
 666 }