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

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