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

   1 /*
   2    Bacula(R) - The Network Backup Solution
   3
   4    Copyright (C) 2000-2015 Kern Sibbald
   5    Copyright (C) 2000-2014 Free Software Foundation Europe e.V.
   6
   7    The original author of Bacula is Kern Sibbald, with contributions
   8    from many others, a complete list can be found in the file AUTHORS.
   9
  10    You may use this file and others of this release according to the
  11    license defined in the LICENSE file, which includes the Affero General
  12    Public License, v3.0 ("AGPLv3") and some additional permissions and
  13    terms pursuant to its AGPLv3 Section 7.
  14
  15    This notice must be preserved when any source code is
  16    conveyed and/or propagated.
  17
  18    Bacula(R) is a registered trademark of Kern Sibbald.
  19 */
  20 /*
  21  * Manipulation routines for Job Control Records and
  22  *  handling of last_jobs_list.
  23  *
  24  *  Kern E. Sibbald, December 2000
  25  *
  26  *  These routines are thread safe.
  27  *
  28  *  The job list routines were re-written in May 2005 to
  29  *  eliminate the global lock while traversing the list, and
  30  *  to use the dlist subroutines.  The locking is now done
  31  *  on the list each time the list is modified or traversed.
  32  *  That is it is "micro-locked" rather than globally locked.
  33  *  The result is that there is one lock/unlock for each entry
  34  *  in the list while traversing it rather than a single lock
  35  *  at the beginning of a traversal and one at the end.  This
  36  *  incurs slightly more overhead, but effectively eliminates
  37  *  the possibilty of race conditions.  In addition, with the
  38  *  exception of the global locking of the list during the
  39  *  re-reading of the config file, no recursion is needed.
  40  *
  41  */
  42
  43 #include "bacula.h"
  44 #include "jcr.h"
  45
  46 const int dbglvl = 3400;
  47
  48 /* External variables we reference */
  49
  50 /* External referenced functions */
  51 void free_bregexps(alist *bregexps);
  52
  53 /* Forward referenced functions */
  54 extern "C" void timeout_handler(int sig);
  55 static void jcr_timeout_check(watchdog_t *self);
  56 #ifdef TRACE_JCR_CHAIN
  57 static void b_lock_jcr_chain(const char *filen, int line);
  58 static void b_unlock_jcr_chain(const char *filen, int line);
  59 #define lock_jcr_chain() b_lock_jcr_chain(__FILE__, __LINE__);
  60 #define unlock_jcr_chain() b_unlock_jcr_chain(__FILE__, __LINE__);
  61 #else
  62 static void lock_jcr_chain();
  63 static void unlock_jcr_chain();
  64 #endif
  65
  66
  67 int num_jobs_run;
  68 dlist *last_jobs = NULL;
  69 const int max_last_jobs = 10;
  70
  71 static dlist *jcrs = NULL;            /* JCR chain */
  72 static pthread_mutex_t jcr_lock = PTHREAD_MUTEX_INITIALIZER;
  73
  74 static pthread_mutex_t job_start_mutex = PTHREAD_MUTEX_INITIALIZER;
  75
  76 static pthread_mutex_t last_jobs_mutex = PTHREAD_MUTEX_INITIALIZER;
  77
  78 static pthread_key_t jcr_key;         /* Pointer to jcr for each thread */
  79
  80 pthread_once_t key_once = PTHREAD_ONCE_INIT;
  81
  82 static char Job_status[]     = "Status Job=%s JobStatus=%d\n";
  83
  84
  85 void lock_jobs()
  86 {
  87    P(job_start_mutex);
  88 }
  89
  90 void unlock_jobs()
  91 {
  92    V(job_start_mutex);
  93 }
  94
  95 void init_last_jobs_list()
  96 {
  97    JCR *jcr = NULL;
  98    struct s_last_job *job_entry = NULL;
  99    if (!last_jobs) {
 100       last_jobs = New(dlist(job_entry, &job_entry->link));
 101    }
 102    if (!jcrs) {
 103       jcrs = New(dlist(jcr, &jcr->link));
 104    }
 105 }
 106
 107 void term_last_jobs_list()
 108 {
 109    if (last_jobs) {
 110       lock_last_jobs_list();
 111       while (!last_jobs->empty()) {
 112          void *je = last_jobs->first();
 113          last_jobs->remove(je);
 114          free(je);
 115       }
 116       delete last_jobs;
 117       last_jobs = NULL;
 118       unlock_last_jobs_list();
 119    }
 120    if (jcrs) {
 121       delete jcrs;
 122       jcrs = NULL;
 123    }
 124 }
 125
 126 bool read_last_jobs_list(int fd, uint64_t addr)
 127 {
 128    struct s_last_job *je, job;
 129    uint32_t num;
 130    bool ok = true;
 131
 132    Dmsg1(100, "read_last_jobs seek to %d\n", (int)addr);
 133    if (addr == 0 || lseek(fd, (boffset_t)addr, SEEK_SET) < 0) {
 134       return false;
 135    }
 136    if (read(fd, &num, sizeof(num)) != sizeof(num)) {
 137       return false;
 138    }
 139    Dmsg1(100, "Read num_items=%d\n", num);
 140    if (num > 4 * max_last_jobs) {  /* sanity check */
 141       return false;
 142    }
 143    lock_last_jobs_list();
 144    for ( ; num; num--) {
 145       if (read(fd, &job, sizeof(job)) != sizeof(job)) {
 146          berrno be;
 147          Pmsg1(000, "Read job entry. ERR=%s\n", be.bstrerror());
 148          ok = false;
 149          break;
 150       }
 151       if (job.JobId > 0) {
 152          je = (struct s_last_job *)malloc(sizeof(struct s_last_job));
 153          memcpy((char *)je, (char *)&job, sizeof(job));
 154          if (!last_jobs) {
 155             init_last_jobs_list();
 156          }
 157          last_jobs->append(je);
 158          if (last_jobs->size() > max_last_jobs) {
 159             je = (struct s_last_job *)last_jobs->first();
 160             last_jobs->remove(je);
 161             free(je);
 162          }
 163       }
 164    }
 165    unlock_last_jobs_list();
 166    return ok;
 167 }
 168
 169 uint64_t write_last_jobs_list(int fd, uint64_t addr)
 170 {
 171    struct s_last_job *je;
 172    uint32_t num;
 173    ssize_t stat;
 174
 175    Dmsg1(100, "write_last_jobs seek to %d\n", (int)addr);
 176    if (lseek(fd, (boffset_t)addr, SEEK_SET) < 0) {
 177       return 0;
 178    }
 179    if (last_jobs) {
 180       lock_last_jobs_list();
 181       /* First record is number of entires */
 182       num = last_jobs->size();
 183       if (write(fd, &num, sizeof(num)) != sizeof(num)) {
 184          berrno be;
 185          Pmsg1(000, "Error writing num_items: ERR=%s\n", be.bstrerror());
 186          goto bail_out;
 187       }
 188       foreach_dlist(je, last_jobs) {
 189          if (write(fd, je, sizeof(struct s_last_job)) != sizeof(struct s_last_job)) {
 190             berrno be;
 191             Pmsg1(000, "Error writing job: ERR=%s\n", be.bstrerror());
 192             goto bail_out;
 193          }
 194       }
 195       unlock_last_jobs_list();
 196    }
 197    /* Return current address */
 198    stat = lseek(fd, 0, SEEK_CUR);
 199    if (stat < 0) {
 200       stat = 0;
 201    }
 202    return stat;
 203
 204 bail_out:
 205    unlock_last_jobs_list();
 206    return 0;
 207 }
 208
 209 void lock_last_jobs_list()
 210 {
 211    P(last_jobs_mutex);
 212 }
 213
 214 void unlock_last_jobs_list()
 215 {
 216    V(last_jobs_mutex);
 217 }
 218
 219 /* Get an ASCII representation of the Operation being performed as an english Noun */
 220 const char *JCR::get_OperationName()
 221 {
 222    switch(m_JobType) {
 223    case JT_BACKUP:
 224       return _("Backup");
 225    case JT_VERIFY:
 226       return _("Verifying");
 227    case JT_RESTORE:
 228       return _("Restoring");
 229    case JT_ARCHIVE:
 230       return _("Archiving");
 231    case JT_COPY:
 232       return _("Copying");
 233    case JT_MIGRATE:
 234       return _("Migration");
 235    case JT_SCAN:
 236       return _("Scanning");
 237    default:
 238       return _("Unknown operation");
 239    }
 240 }
 241
 242 /* Get an ASCII representation of the Action being performed either an english Verb or Adjective */
 243 const char *JCR::get_ActionName(bool past)
 244 {
 245    switch(m_JobType) {
 246    case JT_BACKUP:
 247       return _("backup");
 248    case JT_VERIFY:
 249       return (past == true) ? _("verified") : _("verify");
 250    case JT_RESTORE:
 251       return (past == true) ? _("restored") : _("restore");
 252    case JT_ARCHIVE:
 253       return (past == true) ? _("archived") : _("archive");
 254    case JT_COPY:
 255       return (past == true) ? _("copied") : _("copy");
 256    case JT_MIGRATE:
 257       return (past == true) ? _("migrated") : _("migrate");
 258    case JT_SCAN:
 259       return (past == true) ? _("scanned") : _("scan");
 260    default:
 261       return _("unknown action");
 262    }
 263 }
 264
 265 bool JCR::JobReads()
 266 {
 267    switch (m_JobType) {
 268    case JT_VERIFY:
 269    case JT_RESTORE:
 270    case JT_COPY:
 271    case JT_MIGRATE:
 272       return true;
 273    case JT_BACKUP:
 274       if (m_JobLevel == L_VIRTUAL_FULL) {
 275          return true;
 276       }
 277       break;
 278    default:
 279       break;
 280    }
 281    return false;
 282 }
 283
 284 /*
 285  * Push a subroutine address into the job end callback stack
 286  */
 287 void job_end_push(JCR *jcr, void job_end_cb(JCR *jcr,void *), void *ctx)
 288 {
 289    jcr->job_end_push.append((void *)job_end_cb);
 290    jcr->job_end_push.append(ctx);
 291 }
 292
 293 /* Pop each job_end subroutine and call it */
 294 static void job_end_pop(JCR *jcr)
 295 {
 296    void (*job_end_cb)(JCR *jcr, void *ctx);
 297    void *ctx;
 298    for (int i=jcr->job_end_push.size()-1; i > 0; ) {
 299       ctx = jcr->job_end_push.get(i--);
 300       job_end_cb = (void (*)(JCR *,void *))jcr->job_end_push.get(i--);
 301       job_end_cb(jcr, ctx);
 302    }
 303 }
 304
 305 /*
 306  * Create thread key for thread specific data
 307  */
 308 void create_jcr_key()
 309 {
 310    int status = pthread_key_create(&jcr_key, NULL);
 311    if (status != 0) {
 312       berrno be;
 313       Jmsg1(NULL, M_ABORT, 0, _("pthread key create failed: ERR=%s\n"),
 314             be.bstrerror(status));
 315    }
 316 }
 317
 318 /*
 319  * Create a Job Control Record and link it into JCR chain
 320  * Returns newly allocated JCR
 321  * Note, since each daemon has a different JCR, he passes
 322  *  us the size.
 323  */
 324 JCR *new_jcr(int size, JCR_free_HANDLER *daemon_free_jcr)
 325 {
 326    JCR *jcr;
 327    MQUEUE_ITEM *item = NULL;
 328    int status;
 329
 330    Dmsg0(dbglvl, "Enter new_jcr\n");
 331    status = pthread_once(&key_once, create_jcr_key);
 332    if (status != 0) {
 333       berrno be;
 334       Jmsg1(NULL, M_ABORT, 0, _("pthread_once failed. ERR=%s\n"), be.bstrerror(status));
 335    }
 336    jcr = (JCR *)malloc(size);
 337    memset(jcr, 0, size);
 338    /* Note for the director, this value is changed in jobq.c */
 339    jcr->my_thread_id = pthread_self();
 340    jcr->msg_queue = New(dlist(item, &item->link));
 341    if ((status = pthread_mutex_init(&jcr->msg_queue_mutex, NULL)) != 0) {
 342       berrno be;
 343       Jmsg(NULL, M_ABORT, 0, _("Could not init msg_queue mutex. ERR=%s\n"),
 344          be.bstrerror(status));
 345    }
 346    jcr->job_end_push.init(1, false);
 347    jcr->sched_time = time(NULL);
 348    jcr->initial_sched_time = jcr->sched_time;
 349    jcr->daemon_free_jcr = daemon_free_jcr;    /* plug daemon free routine */
 350    jcr->init_mutex();
 351    jcr->inc_use_count();
 352    jcr->VolumeName = get_pool_memory(PM_FNAME);
 353    jcr->VolumeName[0] = 0;
 354    jcr->errmsg = get_pool_memory(PM_MESSAGE);
 355    jcr->errmsg[0] = 0;
 356    jcr->comment = get_pool_memory(PM_FNAME);
 357    jcr->comment[0] = 0;
 358    /* Setup some dummy values */
 359    bstrncpy(jcr->Job, "*System*", sizeof(jcr->Job));
 360    jcr->JobId = 0;
 361    jcr->setJobType(JT_SYSTEM);           /* internal job until defined */
 362    jcr->setJobLevel(L_NONE);
 363    jcr->setJobStatus(JS_Created);        /* ready to run */
 364 #ifndef HAVE_WIN32
 365    struct sigaction sigtimer;
 366    sigtimer.sa_flags = 0;
 367    sigtimer.sa_handler = timeout_handler;
 368    sigfillset(&sigtimer.sa_mask);
 369    sigaction(TIMEOUT_SIGNAL, &sigtimer, NULL);
 370 #endif
 371
 372    /*
 373     * Locking jobs is a global lock that is needed
 374     * so that the Director can stop new jobs from being
 375     * added to the jcr chain while it processes a new
 376     * conf file and does the job_end_push().
 377     */
 378    lock_jobs();
 379    lock_jcr_chain();
 380    if (!jcrs) {
 381       jcrs = New(dlist(jcr, &jcr->link));
 382    }
 383    jcrs->append(jcr);
 384    unlock_jcr_chain();
 385    unlock_jobs();
 386
 387    return jcr;
 388 }
 389
 390
 391 /*
 392  * Remove a JCR from the chain
 393  * NOTE! The chain must be locked prior to calling
 394  *       this routine.
 395  */
 396 static void remove_jcr(JCR *jcr)
 397 {
 398    Dmsg0(dbglvl, "Enter remove_jcr\n");
 399    if (!jcr) {
 400       Emsg0(M_ABORT, 0, _("NULL jcr.\n"));
 401    }
 402    jcrs->remove(jcr);
 403    Dmsg0(dbglvl, "Leave remove_jcr\n");
 404 }
 405
 406 /*
 407  * Free stuff common to all JCRs.  N.B. Be careful to include only
 408  *  generic stuff in the common part of the jcr.
 409  */
 410 static void free_common_jcr(JCR *jcr)
 411 {
 412    /* Uses jcr lock/unlock */
 413    remove_jcr_from_tsd(jcr);
 414    jcr->set_killable(false);
 415
 416    jcr->destroy_mutex();
 417
 418    if (jcr->msg_queue) {
 419       delete jcr->msg_queue;
 420       jcr->msg_queue = NULL;
 421       pthread_mutex_destroy(&jcr->msg_queue_mutex);
 422    }
 423
 424    /* do this after closing messages */
 425    if (jcr->client_name) {
 426       free_pool_memory(jcr->client_name);
 427       jcr->client_name = NULL;
 428    }
 429
 430    if (jcr->attr) {
 431       free_pool_memory(jcr->attr);
 432       jcr->attr = NULL;
 433    }
 434
 435    if (jcr->sd_auth_key) {
 436       free(jcr->sd_auth_key);
 437       jcr->sd_auth_key = NULL;
 438    }
 439    if (jcr->VolumeName) {
 440       free_pool_memory(jcr->VolumeName);
 441       jcr->VolumeName = NULL;
 442    }
 443
 444    free_bsock(jcr->dir_bsock);
 445
 446    if (jcr->errmsg) {
 447       free_pool_memory(jcr->errmsg);
 448       jcr->errmsg = NULL;
 449    }
 450    if (jcr->where) {
 451       free(jcr->where);
 452       jcr->where = NULL;
 453    }
 454    if (jcr->RegexWhere) {
 455       free(jcr->RegexWhere);
 456       jcr->RegexWhere = NULL;
 457    }
 458    if (jcr->where_bregexp) {
 459       free_bregexps(jcr->where_bregexp);
 460       delete jcr->where_bregexp;
 461       jcr->where_bregexp = NULL;
 462    }
 463    if (jcr->cached_path) {
 464       free_pool_memory(jcr->cached_path);
 465       jcr->cached_path = NULL;
 466       jcr->cached_pnl = 0;
 467    }
 468    if (jcr->id_list) {
 469       free_guid_list(jcr->id_list);
 470       jcr->id_list = NULL;
 471    }
 472    if (jcr->comment) {
 473       free_pool_memory(jcr->comment);
 474       jcr->comment = NULL;
 475    }
 476    free(jcr);
 477 }
 478
 479 /*
 480  * Global routine to free a jcr
 481  */
 482 #ifdef DEBUG
 483 void b_free_jcr(const char *file, int line, JCR *jcr)
 484 {
 485    struct s_last_job *je;
 486
 487    Dmsg3(dbglvl, "Enter free_jcr jid=%u from %s:%d\n", jcr->JobId, file, line);
 488
 489 #else
 490
 491 void free_jcr(JCR *jcr)
 492 {
 493    struct s_last_job *je;
 494
 495    Dmsg3(dbglvl, "Enter free_jcr jid=%u use_count=%d Job=%s\n",
 496          jcr->JobId, jcr->use_count(), jcr->Job);
 497
 498 #endif
 499
 500    lock_jcr_chain();
 501    jcr->dec_use_count();              /* decrement use count */
 502    ASSERT2(jcr->use_count() >= 0, "JCR use_count < 0");
 503    //    Jmsg2(jcr, M_ERROR, 0, _("JCR use_count=%d JobId=%d\n"),
 504    //      jcr->use_count(), jcr->JobId);
 505    //}
 506    if (jcr->JobId > 0) {
 507       Dmsg3(dbglvl, "Dec free_jcr jid=%u use_count=%d Job=%s\n",
 508          jcr->JobId, jcr->use_count(), jcr->Job);
 509    }
 510    if (jcr->use_count() > 0) {          /* if in use */
 511       unlock_jcr_chain();
 512       return;
 513    }
 514    if (jcr->JobId > 0) {
 515       Dmsg3(dbglvl, "remove jcr jid=%u use_count=%d Job=%s\n",
 516             jcr->JobId, jcr->use_count(), jcr->Job);
 517    }
 518    jcr->exiting = true;
 519    remove_jcr(jcr);                   /* remove Jcr from chain */
 520    unlock_jcr_chain();
 521
 522    dequeue_messages(jcr);
 523    close_msg(jcr);                    /* close messages for this job */
 524    job_end_pop(jcr);                  /* pop and call hooked routines */
 525
 526    Dmsg1(dbglvl, "End job=%d\n", jcr->JobId);
 527
 528    /* Keep some statistics */
 529    switch (jcr->getJobType()) {
 530    case JT_BACKUP:
 531    case JT_VERIFY:
 532    case JT_RESTORE:
 533    case JT_MIGRATE:
 534    case JT_COPY:
 535    case JT_ADMIN:
 536       /* Keep list of last jobs, but not Console where JobId==0 */
 537       if (jcr->JobId > 0) {
 538          lock_last_jobs_list();
 539          num_jobs_run++;
 540          je = (struct s_last_job *)malloc(sizeof(struct s_last_job));
 541          memset(je, 0, sizeof(struct s_last_job));  /* zero in case unset fields */
 542          je->Errors = jcr->JobErrors;
 543          je->JobType = jcr->getJobType();
 544          je->JobId = jcr->JobId;
 545          je->VolSessionId = jcr->VolSessionId;
 546          je->VolSessionTime = jcr->VolSessionTime;
 547          bstrncpy(je->Job, jcr->Job, sizeof(je->Job));
 548          je->JobFiles = jcr->JobFiles;
 549          je->JobBytes = jcr->JobBytes;
 550          je->JobStatus = jcr->JobStatus;
 551          je->JobLevel = jcr->getJobLevel();
 552          je->start_time = jcr->start_time;
 553          je->end_time = time(NULL);
 554
 555          if (!last_jobs) {
 556             init_last_jobs_list();
 557          }
 558          last_jobs->append(je);
 559          if (last_jobs->size() > max_last_jobs) {
 560             je = (struct s_last_job *)last_jobs->first();
 561             last_jobs->remove(je);
 562             free(je);
 563          }
 564          unlock_last_jobs_list();
 565       }
 566       break;
 567    default:
 568       break;
 569    }
 570
 571    if (jcr->daemon_free_jcr) {
 572       jcr->daemon_free_jcr(jcr);      /* call daemon free routine */
 573    }
 574
 575    free_common_jcr(jcr);
 576    close_msg(NULL);                   /* flush any daemon messages */
 577    Dmsg0(dbglvl, "Exit free_jcr\n");
 578 }
 579
 580 /*
 581  * Remove jcr from thread specific data, but
 582  *   but make sure it is us who are attached.
 583  */
 584 void remove_jcr_from_tsd(JCR *jcr)
 585 {
 586    JCR *tjcr = get_jcr_from_tsd();
 587    if (tjcr == jcr) {
 588       set_jcr_in_tsd(INVALID_JCR);
 589    }
 590 }
 591
 592 void JCR::set_killable(bool killable)
 593 {
 594    lock();
 595    my_thread_killable = killable;
 596    unlock();
 597 }
 598
 599 /*
 600  * Put this jcr in the thread specifc data
 601  *  if update_thread_info is true and the jcr is valide,
 602  *  we update the my_thread_id in the JCR
 603  */
 604 void set_jcr_in_tsd(JCR *jcr)
 605 {
 606    int status = pthread_setspecific(jcr_key, (void *)jcr);
 607    if (status != 0) {
 608       berrno be;
 609       Jmsg1(jcr, M_ABORT, 0, _("pthread_setspecific failed: ERR=%s\n"),
 610             be.bstrerror(status));
 611    }
 612 }
 613
 614 void JCR::my_thread_send_signal(int sig)
 615 {
 616    lock_jcr_chain();   /* use global lock */
 617    this->lock();
 618    if (this->exiting) {
 619       goto get_out;
 620    }
 621    if (this->is_killable() &&
 622        !pthread_equal(this->my_thread_id, pthread_self()))
 623    {
 624       Dmsg1(800, "Send kill to jid=%d\n", this->JobId);
 625       pthread_kill(this->my_thread_id, sig);
 626       this->exiting = true;
 627
 628    } else if (!this->is_killable()) {
 629       Dmsg1(10, "Warning, can't send kill to jid=%d\n", this->JobId);
 630    }
 631 get_out:
 632    this->unlock();
 633    unlock_jcr_chain();
 634 }
 635
 636 /*
 637  * Give me the jcr that is attached to this thread
 638  */
 639 JCR *get_jcr_from_tsd()
 640 {
 641    JCR *jcr = (JCR *)pthread_getspecific(jcr_key);
 642 // printf("get_jcr_from_tsd: jcr=%p\n", jcr);
 643    /* set any INVALID_JCR to NULL which the rest of Bacula understands */
 644    if (jcr == INVALID_JCR) {
 645       jcr = NULL;
 646    }
 647    return jcr;
 648 }
 649
 650
 651 /*
 652  * Find which JobId corresponds to the current thread
 653  */
 654 uint32_t get_jobid_from_tsd()
 655 {
 656    JCR *jcr;
 657    uint32_t JobId = 0;
 658    jcr = get_jcr_from_tsd();
 659 // printf("get_jobid_from_tsr: jcr=%p\n", jcr);
 660    if (jcr) {
 661       JobId = (uint32_t)jcr->JobId;
 662    }
 663    return JobId;
 664 }
 665
 666 /*
 667  * Given a JobId, find the JCR
 668  *   Returns: jcr on success
 669  *            NULL on failure
 670  */
 671 JCR *get_jcr_by_id(uint32_t JobId)
 672 {
 673    JCR *jcr;
 674
 675    foreach_jcr(jcr) {
 676       if (jcr->JobId == JobId) {
 677          jcr->inc_use_count();
 678          Dmsg3(dbglvl, "Inc get_jcr jid=%u use_count=%d Job=%s\n",
 679             jcr->JobId, jcr->use_count(), jcr->Job);
 680          break;
 681       }
 682    }
 683    endeach_jcr(jcr);
 684    return jcr;
 685 }
 686
 687 /*
 688  * Given a thread id, find the JobId
 689  *   Returns: JobId on success
 690  *            0 on failure
 691  */
 692 uint32_t get_jobid_from_tid(pthread_t tid)
 693 {
 694    JCR *jcr = NULL;
 695    bool found = false;
 696
 697    foreach_jcr(jcr) {
 698       if (pthread_equal(jcr->my_thread_id, tid)) {
 699          found = true;
 700          break;
 701       }
 702    }
 703    endeach_jcr(jcr);
 704    if (found) {
 705       return jcr->JobId;
 706    }
 707    return 0;
 708 }
 709
 710
 711 /*
 712  * Given a SessionId and SessionTime, find the JCR
 713  *   Returns: jcr on success
 714  *            NULL on failure
 715  */
 716 JCR *get_jcr_by_session(uint32_t SessionId, uint32_t SessionTime)
 717 {
 718    JCR *jcr;
 719
 720    foreach_jcr(jcr) {
 721       if (jcr->VolSessionId == SessionId &&
 722           jcr->VolSessionTime == SessionTime) {
 723          jcr->inc_use_count();
 724          Dmsg3(dbglvl, "Inc get_jcr jid=%u use_count=%d Job=%s\n",
 725             jcr->JobId, jcr->use_count(), jcr->Job);
 726          break;
 727       }
 728    }
 729    endeach_jcr(jcr);
 730    return jcr;
 731 }
 732
 733
 734 /*
 735  * Given a Job, find the JCR
 736  *  compares on the number of characters in Job
 737  *  thus allowing partial matches.
 738  *   Returns: jcr on success
 739  *            NULL on failure
 740  */
 741 JCR *get_jcr_by_partial_name(char *Job)
 742 {
 743    JCR *jcr;
 744    int len;
 745
 746    if (!Job) {
 747       return NULL;
 748    }
 749    len = strlen(Job);
 750    foreach_jcr(jcr) {
 751       if (strncmp(Job, jcr->Job, len) == 0) {
 752          jcr->inc_use_count();
 753          Dmsg3(dbglvl, "Inc get_jcr jid=%u use_count=%d Job=%s\n",
 754             jcr->JobId, jcr->use_count(), jcr->Job);
 755          break;
 756       }
 757    }
 758    endeach_jcr(jcr);
 759    return jcr;
 760 }
 761
 762
 763 /*
 764  * Given a Job, find the JCR
 765  *  requires an exact match of names.
 766  *   Returns: jcr on success
 767  *            NULL on failure
 768  */
 769 JCR *get_jcr_by_full_name(char *Job)
 770 {
 771    JCR *jcr;
 772
 773    if (!Job) {
 774       return NULL;
 775    }
 776    foreach_jcr(jcr) {
 777       if (strcmp(jcr->Job, Job) == 0) {
 778          jcr->inc_use_count();
 779          Dmsg3(dbglvl, "Inc get_jcr jid=%u use_count=%d Job=%s\n",
 780             jcr->JobId, jcr->use_count(), jcr->Job);
 781          break;
 782       }
 783    }
 784    endeach_jcr(jcr);
 785    return jcr;
 786 }
 787
 788 static void update_wait_time(JCR *jcr, int newJobStatus)
 789 {
 790    bool enter_in_waittime;
 791    int oldJobStatus = jcr->JobStatus;
 792
 793    switch (newJobStatus) {
 794    case JS_WaitFD:
 795    case JS_WaitSD:
 796    case JS_WaitMedia:
 797    case JS_WaitMount:
 798    case JS_WaitStoreRes:
 799    case JS_WaitJobRes:
 800    case JS_WaitClientRes:
 801    case JS_WaitMaxJobs:
 802    case JS_WaitPriority:
 803       enter_in_waittime = true;
 804       break;
 805    default:
 806       enter_in_waittime = false; /* not a Wait situation */
 807       break;
 808    }
 809
 810    /*
 811     * If we were previously waiting and are not any more
 812     *   we want to update the wait_time variable, which is
 813     *   the start of waiting.
 814     */
 815    switch (oldJobStatus) {
 816    case JS_WaitFD:
 817    case JS_WaitSD:
 818    case JS_WaitMedia:
 819    case JS_WaitMount:
 820    case JS_WaitStoreRes:
 821    case JS_WaitJobRes:
 822    case JS_WaitClientRes:
 823    case JS_WaitMaxJobs:
 824    case JS_WaitPriority:
 825       if (!enter_in_waittime) { /* we get out the wait time */
 826          jcr->wait_time_sum += (time(NULL) - jcr->wait_time);
 827          jcr->wait_time = 0;
 828       }
 829       break;
 830
 831    /* if wait state is new, we keep current time for watchdog MaxWaitTime */
 832    default:
 833       if (enter_in_waittime) {
 834          jcr->wait_time = time(NULL);
 835       }
 836       break;
 837    }
 838 }
 839
 840 /*
 841  * Priority runs from 0 (lowest) to 10 (highest)
 842  */
 843 static int get_status_priority(int JobStatus)
 844 {
 845    int priority = 0;
 846    switch (JobStatus) {
 847    case JS_Incomplete:
 848       priority = 10;
 849       break;
 850    case JS_ErrorTerminated:
 851    case JS_FatalError:
 852    case JS_Canceled:
 853       priority = 9;
 854       break;
 855    case JS_Error:
 856       priority = 8;
 857       break;
 858    case JS_Differences:
 859       priority = 7;
 860       break;
 861    }
 862    return priority;
 863 }
 864
 865 /*
 866  * Send Job status to Director
 867  */
 868 bool JCR::sendJobStatus()
 869 {
 870    if (dir_bsock) {
 871       return dir_bsock->fsend(Job_status, Job, JobStatus);
 872    }
 873    return true;
 874 }
 875
 876 /*
 877  * Set and send Job status to Director
 878  */
 879 bool JCR::sendJobStatus(int aJobStatus)
 880 {
 881    if (!is_JobStatus(aJobStatus)) {
 882       setJobStatus(aJobStatus);
 883       if (dir_bsock) {
 884          return dir_bsock->fsend(Job_status, Job, JobStatus);
 885       }
 886    }
 887    return true;
 888 }
 889
 890 void JCR::setJobStarted()
 891 {
 892    job_started = true;
 893    job_started_time = time(NULL);
 894 }
 895
 896 void JCR::setJobStatus(int newJobStatus)
 897 {
 898    int priority, old_priority;
 899    int oldJobStatus = JobStatus;
 900    priority = get_status_priority(newJobStatus);
 901    old_priority = get_status_priority(oldJobStatus);
 902
 903    Dmsg2(800, "set_jcr_job_status(%s, %c)\n", Job, newJobStatus);
 904
 905    /* Update wait_time depending on newJobStatus and oldJobStatus */
 906    update_wait_time(this, newJobStatus);
 907
 908    /*
 909     * For a set of errors, ... keep the current status
 910     *   so it isn't lost. For all others, set it.
 911     */
 912    Dmsg2(800, "OnEntry JobStatus=%c newJobstatus=%c\n", oldJobStatus, newJobStatus);
 913    /*
 914     * If status priority is > than proposed new status, change it.
 915     * If status priority == new priority and both are zero, take
 916     *   the new status.
 917     * If it is not zero, then we keep the first non-zero "error" that
 918     *   occurred.
 919     */
 920    if (priority > old_priority || (
 921        priority == 0 && old_priority == 0)) {
 922       Dmsg4(800, "Set new stat. old: %c,%d new: %c,%d\n",
 923          JobStatus, old_priority, newJobStatus, priority);
 924       JobStatus = newJobStatus;     /* replace with new status */
 925    }
 926
 927    if (oldJobStatus != JobStatus) {
 928       Dmsg2(800, "leave setJobStatus old=%c new=%c\n", oldJobStatus, newJobStatus);
 929 //    generate_plugin_event(this, bEventStatusChange, NULL);
 930    }
 931 }
 932
 933 #ifdef TRACE_JCR_CHAIN
 934 static int lock_count = 0;
 935 #endif
 936
 937 /*
 938  * Lock the chain
 939  */
 940 #ifdef TRACE_JCR_CHAIN
 941 static void b_lock_jcr_chain(const char *fname, int line)
 942 #else
 943 static void lock_jcr_chain()
 944 #endif
 945 {
 946 #ifdef TRACE_JCR_CHAIN
 947    Dmsg3(dbglvl, "Lock jcr chain %d from %s:%d\n", ++lock_count, fname, line);
 948 #endif
 949    P(jcr_lock);
 950 }
 951
 952 /*
 953  * Unlock the chain
 954  */
 955 #ifdef TRACE_JCR_CHAIN
 956 static void b_unlock_jcr_chain(const char *fname, int line)
 957 #else
 958 static void unlock_jcr_chain()
 959 #endif
 960 {
 961 #ifdef TRACE_JCR_CHAIN
 962    Dmsg3(dbglvl, "Unlock jcr chain %d from %s:%d\n", lock_count--, fname, line);
 963 #endif
 964    V(jcr_lock);
 965 }
 966
 967 /*
 968  * Start walk of jcr chain
 969  * The proper way to walk the jcr chain is:
 970  *    JCR *jcr;
 971  *    foreach_jcr(jcr) {
 972  *      ...
 973  *    }
 974  *    endeach_jcr(jcr);
 975  *
 976  *  It is possible to leave out the endeach_jcr(jcr), but
 977  *   in that case, the last jcr referenced must be explicitly
 978  *   released with:
 979  *
 980  *    free_jcr(jcr);
 981  *
 982  */
 983 JCR *jcr_walk_start()
 984 {
 985    JCR *jcr;
 986    lock_jcr_chain();
 987    jcr = (JCR *)jcrs->first();
 988    if (jcr) {
 989       jcr->inc_use_count();
 990       if (jcr->JobId > 0) {
 991          Dmsg3(dbglvl, "Inc walk_start jid=%u use_count=%d Job=%s\n",
 992             jcr->JobId, jcr->use_count(), jcr->Job);
 993       }
 994    }
 995    unlock_jcr_chain();
 996    return jcr;
 997 }
 998
 999 /*
1000  * Get next jcr from chain, and release current one
1001  */
1002 JCR *jcr_walk_next(JCR *prev_jcr)
1003 {
1004    JCR *jcr;
1005
1006    lock_jcr_chain();
1007    jcr = (JCR *)jcrs->next(prev_jcr);
1008    if (jcr) {
1009       jcr->inc_use_count();
1010       if (jcr->JobId > 0) {
1011          Dmsg3(dbglvl, "Inc walk_next jid=%u use_count=%d Job=%s\n",
1012             jcr->JobId, jcr->use_count(), jcr->Job);
1013       }
1014    }
1015    unlock_jcr_chain();
1016    if (prev_jcr) {
1017       free_jcr(prev_jcr);
1018    }
1019    return jcr;
1020 }
1021
1022 /*
1023  * Release last jcr referenced
1024  */
1025 void jcr_walk_end(JCR *jcr)
1026 {
1027    if (jcr) {
1028       if (jcr->JobId > 0) {
1029          Dmsg3(dbglvl, "Free walk_end jid=%u use_count=%d Job=%s\n",
1030             jcr->JobId, jcr->use_count(), jcr->Job);
1031       }
1032       free_jcr(jcr);
1033    }
1034 }
1035
1036 /*
1037  * Return number of Jobs
1038  */
1039 int job_count()
1040 {
1041    JCR *jcr;
1042    int count = 0;
1043
1044    lock_jcr_chain();
1045    for (jcr = (JCR *)jcrs->first(); (jcr = (JCR *)jcrs->next(jcr)); ) {
1046       if (jcr->JobId > 0) {
1047          count++;
1048       }
1049    }
1050    unlock_jcr_chain();
1051    return count;
1052 }
1053
1054
1055 /*
1056  * Setup to call the timeout check routine every 30 seconds
1057  *  This routine will check any timers that have been enabled.
1058  */
1059 bool init_jcr_subsystem(void)
1060 {
1061    watchdog_t *wd = new_watchdog();
1062
1063    wd->one_shot = false;
1064    wd->interval = 30;   /* FIXME: should be configurable somewhere, even
1065                          if only with a #define */
1066    wd->callback = jcr_timeout_check;
1067
1068    register_watchdog(wd);
1069
1070    return true;
1071 }
1072
1073 static void jcr_timeout_check(watchdog_t *self)
1074 {
1075    JCR *jcr;
1076    BSOCK *bs;
1077    time_t timer_start;
1078
1079    Dmsg0(dbglvl, "Start JCR timeout checks\n");
1080
1081    /* Walk through all JCRs checking if any one is
1082     * blocked for more than specified max time.
1083     */
1084    foreach_jcr(jcr) {
1085       Dmsg2(dbglvl, "jcr_timeout_check JobId=%u jcr=0x%x\n", jcr->JobId, jcr);
1086       if (jcr->JobId == 0) {
1087          continue;
1088       }
1089       bs = jcr->store_bsock;
1090       if (bs) {
1091          timer_start = bs->timer_start;
1092          if (timer_start && (watchdog_time - timer_start) > bs->timeout) {
1093             bs->timer_start = 0;      /* turn off timer */
1094             bs->set_timed_out();
1095             Qmsg(jcr, M_ERROR, 0, _(
1096 "Watchdog sending kill after %d secs to thread stalled reading Storage daemon.\n"),
1097                  (int)(watchdog_time - timer_start));
1098             jcr->my_thread_send_signal(TIMEOUT_SIGNAL);
1099          }
1100       }
1101       bs = jcr->file_bsock;
1102       if (bs) {
1103          timer_start = bs->timer_start;
1104          if (timer_start && (watchdog_time - timer_start) > bs->timeout) {
1105             bs->timer_start = 0;      /* turn off timer */
1106             bs->set_timed_out();
1107             Qmsg(jcr, M_ERROR, 0, _(
1108 "Watchdog sending kill after %d secs to thread stalled reading File daemon.\n"),
1109                  (int)(watchdog_time - timer_start));
1110             jcr->my_thread_send_signal(TIMEOUT_SIGNAL);
1111          }
1112       }
1113       bs = jcr->dir_bsock;
1114       if (bs) {
1115          timer_start = bs->timer_start;
1116          if (timer_start && (watchdog_time - timer_start) > bs->timeout) {
1117             bs->timer_start = 0;      /* turn off timer */
1118             bs->set_timed_out();
1119             Qmsg(jcr, M_ERROR, 0, _(
1120 "Watchdog sending kill after %d secs to thread stalled reading Director.\n"),
1121                  (int)(watchdog_time - timer_start));
1122             jcr->my_thread_send_signal(TIMEOUT_SIGNAL);
1123          }
1124       }
1125    }
1126    endeach_jcr(jcr);
1127
1128    Dmsg0(dbglvl, "Finished JCR timeout checks\n");
1129 }
1130
1131 /*
1132  * Return next JobId from comma separated list
1133  *
1134  * Returns:
1135  *   1 if next JobId returned
1136  *   0 if no more JobIds are in list
1137  *  -1 there is an error
1138  */
1139 int get_next_jobid_from_list(char **p, uint32_t *JobId)
1140 {
1141    const int maxlen = 30;
1142    char jobid[maxlen+1];
1143    char *q = *p;
1144
1145    jobid[0] = 0;
1146    for (int i=0; i<maxlen; i++) {
1147       if (*q == 0) {
1148          break;
1149       } else if (*q == ',') {
1150          q++;
1151          break;
1152       }
1153       jobid[i] = *q++;
1154       jobid[i+1] = 0;
1155    }
1156    if (jobid[0] == 0) {
1157       return 0;
1158    } else if (!is_a_number(jobid)) {
1159       return -1;                      /* error */
1160    }
1161    *p = q;
1162    *JobId = str_to_int64(jobid);
1163    return 1;
1164 }
1165
1166 /*
1167  * Timeout signal comes here
1168  */
1169 extern "C" void timeout_handler(int sig)
1170 {
1171    return;                            /* thus interrupting the function */
1172 }
1173
1174 /* Used to display specific daemon information after a fatal signal
1175  * (like BDB in the director)
1176  */
1177 #define MAX_DBG_HOOK 10
1178 static dbg_jcr_hook_t *dbg_jcr_hooks[MAX_DBG_HOOK];
1179 static int dbg_jcr_handler_count;
1180
1181 void dbg_jcr_add_hook(dbg_jcr_hook_t *hook)
1182 {
1183    ASSERT(dbg_jcr_handler_count < MAX_DBG_HOOK);
1184    dbg_jcr_hooks[dbg_jcr_handler_count++] = hook;
1185 }
1186
1187 /* on win32, the pthread_t is a struct, so we don't display it */
1188 #ifdef HAVE_MINGW_W64
1189 # define get_threadid(a) (void *)0
1190 #else
1191 # define get_threadid(a) (void *)(a)
1192 #endif
1193 /*
1194  * !!! WARNING !!!
1195  *
1196  * This function should be used ONLY after a fatal signal. We walk through the
1197  * JCR chain without doing any lock, Bacula should not be running.
1198  */
1199 void dbg_print_jcr(FILE *fp)
1200 {
1201    char buf1[128], buf2[128], buf3[128], buf4[128];
1202
1203    if (!jcrs) {
1204       return;
1205    }
1206
1207    fprintf(fp, "Attempt to dump current JCRs. njcrs=%d\n", jcrs->size());
1208
1209    for (JCR *jcr = (JCR *)jcrs->first(); jcr ; jcr = (JCR *)jcrs->next(jcr)) {
1210       fprintf(fp, "threadid=%p JobId=%d JobStatus=%c jcr=%p name=%s\n",
1211               get_threadid(jcr->my_thread_id), (int)jcr->JobId, jcr->JobStatus, jcr, jcr->Job);
1212       fprintf(fp, "\tuse_count=%i killable=%d\n",
1213               jcr->use_count(), jcr->is_killable());
1214       fprintf(fp, "\tJobType=%c JobLevel=%c\n",
1215               jcr->getJobType(), jcr->getJobLevel());
1216       bstrftime(buf1, sizeof(buf1), jcr->sched_time);
1217       bstrftime(buf2, sizeof(buf2), jcr->start_time);
1218       bstrftime(buf3, sizeof(buf3), jcr->end_time);
1219       bstrftime(buf4, sizeof(buf4), jcr->wait_time);
1220       fprintf(fp, "\tsched_time=%s start_time=%s\n\tend_time=%s wait_time=%s\n",
1221               buf1, buf2, buf3, buf4);
1222       fprintf(fp, "\tdb=%p db_batch=%p batch_started=%i\n",
1223               jcr->db, jcr->db_batch, jcr->batch_started);
1224
1225       /*
1226        * Call all the jcr debug hooks
1227        */
1228       for(int i=0; i < dbg_jcr_handler_count; i++) {
1229          dbg_jcr_hook_t *hook = dbg_jcr_hooks[i];
1230          hook(jcr, fp);
1231       }
1232    }
1233 }