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

   1 /*
   2    Bacula® - The Network Backup Solution
   3
   4    Copyright (C) 2003-2008 Free Software Foundation Europe e.V.
   5
   6    The main author of Bacula is Kern Sibbald, with contributions from
   7    many others, a complete list can be found in the file AUTHORS.
   8    This program is Free Software; you can redistribute it and/or
   9    modify it under the terms of version two of the GNU General Public
  10    License as published by the Free Software Foundation and included
  11    in the file LICENSE.
  12
  13    This program is distributed in the hope that it will be useful, but
  14    WITHOUT ANY WARRANTY; without even the implied warranty of
  15    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  16    General Public License for more details.
  17
  18    You should have received a copy of the GNU General Public License
  19    along with this program; if not, write to the Free Software
  20    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
  21    02110-1301, USA.
  22
  23    Bacula® is a registered trademark of Kern Sibbald.
  24    The licensor of Bacula is the Free Software Foundation Europe
  25    (FSFE), Fiduciary Program, Sumatrastrasse 25, 8006 Zürich,
  26    Switzerland, email:ftf@fsfeurope.org.
  27 */
  28 /*
  29  * Bacula job queue routines.
  30  *
  31  *  This code consists of three queues, the waiting_jobs
  32  *  queue, where jobs are initially queued, the ready_jobs
  33  *  queue, where jobs are placed when all the resources are
  34  *  allocated and they can immediately be run, and the
  35  *  running queue where jobs are placed when they are
  36  *  running.
  37  *
  38  *  Kern Sibbald, July MMIII
  39  *
  40  *   Version $Id$
  41  *
  42  *  This code was adapted from the Bacula workq, which was
  43  *    adapted from "Programming with POSIX Threads", by
  44  *    David R. Butenhof
  45  *
  46  */
  47
  48 #include "bacula.h"
  49 #include "dird.h"
  50
  51 extern JCR *jobs;
  52
  53 /* Forward referenced functions */
  54 extern "C" void *jobq_server(void *arg);
  55 extern "C" void *sched_wait(void *arg);
  56
  57 static int  start_server(jobq_t *jq);
  58 static bool acquire_resources(JCR *jcr);
  59 static bool reschedule_job(JCR *jcr, jobq_t *jq, jobq_item_t *je);
  60 static void dec_read_store(JCR *jcr);
  61 static void dec_write_store(JCR *jcr);
  62
  63 /*
  64  * Initialize a job queue
  65  *
  66  *  Returns: 0 on success
  67  *           errno on failure
  68  */
  69 int jobq_init(jobq_t *jq, int threads, void *(*engine)(void *arg))
  70 {
  71    int stat;
  72    jobq_item_t *item = NULL;
  73
  74    if ((stat = pthread_attr_init(&jq->attr)) != 0) {
  75       berrno be;
  76       Jmsg1(NULL, M_ERROR, 0, _("pthread_attr_init: ERR=%s\n"), be.bstrerror(stat));
  77       return stat;
  78    }
  79    if ((stat = pthread_attr_setdetachstate(&jq->attr, PTHREAD_CREATE_DETACHED)) != 0) {
  80       pthread_attr_destroy(&jq->attr);
  81       return stat;
  82    }
  83    if ((stat = pthread_mutex_init(&jq->mutex, NULL)) != 0) {
  84       berrno be;
  85       Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_init: ERR=%s\n"), be.bstrerror(stat));
  86       pthread_attr_destroy(&jq->attr);
  87       return stat;
  88    }
  89    if ((stat = pthread_cond_init(&jq->work, NULL)) != 0) {
  90       berrno be;
  91       Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_init: ERR=%s\n"), be.bstrerror(stat));
  92       pthread_mutex_destroy(&jq->mutex);
  93       pthread_attr_destroy(&jq->attr);
  94       return stat;
  95    }
  96    jq->quit = false;
  97    jq->max_workers = threads;         /* max threads to create */
  98    jq->num_workers = 0;               /* no threads yet */
  99    jq->idle_workers = 0;              /* no idle threads */
 100    jq->engine = engine;               /* routine to run */
 101    jq->valid = JOBQ_VALID;
 102    /* Initialize the job queues */
 103    jq->waiting_jobs = New(dlist(item, &item->link));
 104    jq->running_jobs = New(dlist(item, &item->link));
 105    jq->ready_jobs = New(dlist(item, &item->link));
 106    return 0;
 107 }
 108
 109 /*
 110  * Destroy the job queue
 111  *
 112  * Returns: 0 on success
 113  *          errno on failure
 114  */
 115 int jobq_destroy(jobq_t *jq)
 116 {
 117    int stat, stat1, stat2;
 118
 119    if (jq->valid != JOBQ_VALID) {
 120       return EINVAL;
 121    }
 122    if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 123       berrno be;
 124       Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_lock: ERR=%s\n"), be.bstrerror(stat));
 125       return stat;
 126    }
 127    jq->valid = 0;                      /* prevent any more operations */
 128
 129    /*
 130     * If any threads are active, wake them
 131     */
 132    if (jq->num_workers > 0) {
 133       jq->quit = true;
 134       if (jq->idle_workers) {
 135          if ((stat = pthread_cond_broadcast(&jq->work)) != 0) {
 136             berrno be;
 137             Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_broadcast: ERR=%s\n"), be.bstrerror(stat));
 138             pthread_mutex_unlock(&jq->mutex);
 139             return stat;
 140          }
 141       }
 142       while (jq->num_workers > 0) {
 143          if ((stat = pthread_cond_wait(&jq->work, &jq->mutex)) != 0) {
 144             berrno be;
 145             Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_wait: ERR=%s\n"), be.bstrerror(stat));
 146             pthread_mutex_unlock(&jq->mutex);
 147             return stat;
 148          }
 149       }
 150    }
 151    if ((stat = pthread_mutex_unlock(&jq->mutex)) != 0) {
 152       berrno be;
 153       Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_unlock: ERR=%s\n"), be.bstrerror(stat));
 154       return stat;
 155    }
 156    stat  = pthread_mutex_destroy(&jq->mutex);
 157    stat1 = pthread_cond_destroy(&jq->work);
 158    stat2 = pthread_attr_destroy(&jq->attr);
 159    delete jq->waiting_jobs;
 160    delete jq->running_jobs;
 161    delete jq->ready_jobs;
 162    return (stat != 0 ? stat : (stat1 != 0 ? stat1 : stat2));
 163 }
 164
 165 struct wait_pkt {
 166    JCR *jcr;
 167    jobq_t *jq;
 168 };
 169
 170 /*
 171  * Wait until schedule time arrives before starting. Normally
 172  *  this routine is only used for jobs started from the console
 173  *  for which the user explicitly specified a start time. Otherwise
 174  *  most jobs are put into the job queue only when their
 175  *  scheduled time arives.
 176  */
 177 extern "C"
 178 void *sched_wait(void *arg)
 179 {
 180    JCR *jcr = ((wait_pkt *)arg)->jcr;
 181    jobq_t *jq = ((wait_pkt *)arg)->jq;
 182
 183    Dmsg0(2300, "Enter sched_wait.\n");
 184    free(arg);
 185    time_t wtime = jcr->sched_time - time(NULL);
 186    set_jcr_job_status(jcr, JS_WaitStartTime);
 187    /* Wait until scheduled time arrives */
 188    if (wtime > 0) {
 189       Jmsg(jcr, M_INFO, 0, _("Job %s waiting %d seconds for scheduled start time.\n"),
 190          jcr->Job, wtime);
 191    }
 192    /* Check every 30 seconds if canceled */
 193    while (wtime > 0) {
 194       Dmsg3(2300, "Waiting on sched time, jobid=%d secs=%d use=%d\n",
 195          jcr->JobId, wtime, jcr->use_count());
 196       if (wtime > 30) {
 197          wtime = 30;
 198       }
 199       bmicrosleep(wtime, 0);
 200       if (job_canceled(jcr)) {
 201          break;
 202       }
 203       wtime = jcr->sched_time - time(NULL);
 204    }
 205    Dmsg1(200, "resched use=%d\n", jcr->use_count());
 206    jobq_add(jq, jcr);
 207    free_jcr(jcr);                     /* we are done with jcr */
 208    Dmsg0(2300, "Exit sched_wait\n");
 209    return NULL;
 210 }
 211
 212 /*
 213  *  Add a job to the queue
 214  *    jq is a queue that was created with jobq_init
 215  */
 216 int jobq_add(jobq_t *jq, JCR *jcr)
 217 {
 218    int stat;
 219    jobq_item_t *item, *li;
 220    bool inserted = false;
 221    time_t wtime = jcr->sched_time - time(NULL);
 222    pthread_t id;
 223    wait_pkt *sched_pkt;
 224
 225    if (!jcr->term_wait_inited) {
 226       /* Initialize termination condition variable */
 227       if ((stat = pthread_cond_init(&jcr->term_wait, NULL)) != 0) {
 228          berrno be;
 229          Jmsg1(jcr, M_FATAL, 0, _("Unable to init job cond variable: ERR=%s\n"), be.bstrerror(stat));
 230          return stat;
 231       }
 232       jcr->term_wait_inited = true;
 233    }
 234
 235    Dmsg3(2300, "jobq_add jobid=%d jcr=0x%x use_count=%d\n", jcr->JobId, jcr, jcr->use_count());
 236    if (jq->valid != JOBQ_VALID) {
 237       Jmsg0(jcr, M_ERROR, 0, "Jobq_add queue not initialized.\n");
 238       return EINVAL;
 239    }
 240
 241    jcr->inc_use_count();                 /* mark jcr in use by us */
 242    Dmsg3(2300, "jobq_add jobid=%d jcr=0x%x use_count=%d\n", jcr->JobId, jcr, jcr->use_count());
 243    if (!job_canceled(jcr) && wtime > 0) {
 244       set_thread_concurrency(jq->max_workers + 2);
 245       sched_pkt = (wait_pkt *)malloc(sizeof(wait_pkt));
 246       sched_pkt->jcr = jcr;
 247       sched_pkt->jq = jq;
 248       stat = pthread_create(&id, &jq->attr, sched_wait, (void *)sched_pkt);
 249       if (stat != 0) {                /* thread not created */
 250          berrno be;
 251          Jmsg1(jcr, M_ERROR, 0, _("pthread_thread_create: ERR=%s\n"), be.bstrerror(stat));
 252       }
 253       return stat;
 254    }
 255
 256    if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 257       berrno be;
 258       Jmsg1(jcr, M_ERROR, 0, _("pthread_mutex_lock: ERR=%s\n"), be.bstrerror(stat));
 259       free_jcr(jcr);                    /* release jcr */
 260       return stat;
 261    }
 262
 263    if ((item = (jobq_item_t *)malloc(sizeof(jobq_item_t))) == NULL) {
 264       free_jcr(jcr);                    /* release jcr */
 265       return ENOMEM;
 266    }
 267    item->jcr = jcr;
 268
 269    if (job_canceled(jcr)) {
 270       /* Add job to ready queue so that it is canceled quickly */
 271       jq->ready_jobs->prepend(item);
 272       Dmsg1(2300, "Prepended job=%d to ready queue\n", jcr->JobId);
 273    } else {
 274       /* Add this job to the wait queue in priority sorted order */
 275       foreach_dlist(li, jq->waiting_jobs) {
 276          Dmsg2(2300, "waiting item jobid=%d priority=%d\n",
 277             li->jcr->JobId, li->jcr->JobPriority);
 278          if (li->jcr->JobPriority > jcr->JobPriority) {
 279             jq->waiting_jobs->insert_before(item, li);
 280             Dmsg2(2300, "insert_before jobid=%d before waiting job=%d\n",
 281                li->jcr->JobId, jcr->JobId);
 282             inserted = true;
 283             break;
 284          }
 285       }
 286       /* If not jobs in wait queue, append it */
 287       if (!inserted) {
 288          jq->waiting_jobs->append(item);
 289          Dmsg1(2300, "Appended item jobid=%d to waiting queue\n", jcr->JobId);
 290       }
 291    }
 292
 293    /* Ensure that at least one server looks at the queue. */
 294    stat = start_server(jq);
 295
 296    pthread_mutex_unlock(&jq->mutex);
 297    Dmsg0(2300, "Return jobq_add\n");
 298    return stat;
 299 }
 300
 301 /*
 302  *  Remove a job from the job queue. Used only by cancel_job().
 303  *    jq is a queue that was created with jobq_init
 304  *    work_item is an element of work
 305  *
 306  *   Note, it is "removed" from the job queue.
 307  *    If you want to cancel it, you need to provide some external means
 308  *    of doing so (e.g. pthread_kill()).
 309  */
 310 int jobq_remove(jobq_t *jq, JCR *jcr)
 311 {
 312    int stat;
 313    bool found = false;
 314    jobq_item_t *item;
 315
 316    Dmsg2(2300, "jobq_remove jobid=%d jcr=0x%x\n", jcr->JobId, jcr);
 317    if (jq->valid != JOBQ_VALID) {
 318       return EINVAL;
 319    }
 320
 321    if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 322       berrno be;
 323       Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_lock: ERR=%s\n"), be.bstrerror(stat));
 324       return stat;
 325    }
 326
 327    foreach_dlist(item, jq->waiting_jobs) {
 328       if (jcr == item->jcr) {
 329          found = true;
 330          break;
 331       }
 332    }
 333    if (!found) {
 334       pthread_mutex_unlock(&jq->mutex);
 335       Dmsg2(2300, "jobq_remove jobid=%d jcr=0x%x not in wait queue\n", jcr->JobId, jcr);
 336       return EINVAL;
 337    }
 338
 339    /* Move item to be the first on the list */
 340    jq->waiting_jobs->remove(item);
 341    jq->ready_jobs->prepend(item);
 342    Dmsg2(2300, "jobq_remove jobid=%d jcr=0x%x moved to ready queue\n", jcr->JobId, jcr);
 343
 344    stat = start_server(jq);
 345
 346    pthread_mutex_unlock(&jq->mutex);
 347    Dmsg0(2300, "Return jobq_remove\n");
 348    return stat;
 349 }
 350
 351
 352 /*
 353  * Start the server thread if it isn't already running
 354  */
 355 static int start_server(jobq_t *jq)
 356 {
 357    int stat = 0;
 358    pthread_t id;
 359
 360    /*
 361     * if any threads are idle, wake one.
 362     *   Actually we do a broadcast because on /lib/tls
 363     *   these signals seem to get lost from time to time.
 364     */
 365    if (jq->idle_workers > 0) {
 366       Dmsg0(2300, "Signal worker to wake up\n");
 367       if ((stat = pthread_cond_broadcast(&jq->work)) != 0) {
 368          berrno be;
 369          Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_signal: ERR=%s\n"), be.bstrerror(stat));
 370          return stat;
 371       }
 372    } else if (jq->num_workers < jq->max_workers) {
 373       Dmsg0(2300, "Create worker thread\n");
 374       /* No idle threads so create a new one */
 375       set_thread_concurrency(jq->max_workers + 1);
 376       if ((stat = pthread_create(&id, &jq->attr, jobq_server, (void *)jq)) != 0) {
 377          berrno be;
 378          Jmsg1(NULL, M_ERROR, 0, _("pthread_create: ERR=%s\n"), be.bstrerror(stat));
 379          return stat;
 380       }
 381    }
 382    return stat;
 383 }
 384
 385
 386 /*
 387  * This is the worker thread that serves the job queue.
 388  * When all the resources are acquired for the job,
 389  *  it will call the user's engine.
 390  */
 391 extern "C"
 392 void *jobq_server(void *arg)
 393 {
 394    struct timespec timeout;
 395    jobq_t *jq = (jobq_t *)arg;
 396    jobq_item_t *je;                   /* job entry in queue */
 397    int stat;
 398    bool timedout = false;
 399    bool work = true;
 400
 401    Dmsg0(2300, "Start jobq_server\n");
 402    if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 403       berrno be;
 404       Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_lock: ERR=%s\n"), be.bstrerror(stat));
 405       return NULL;
 406    }
 407    jq->num_workers++;
 408
 409    for (;;) {
 410       struct timeval tv;
 411       struct timezone tz;
 412
 413       Dmsg0(2300, "Top of for loop\n");
 414       if (!work && !jq->quit) {
 415          gettimeofday(&tv, &tz);
 416          timeout.tv_nsec = 0;
 417          timeout.tv_sec = tv.tv_sec + 4;
 418
 419          while (!jq->quit) {
 420             /*
 421              * Wait 4 seconds, then if no more work, exit
 422              */
 423             Dmsg0(2300, "pthread_cond_timedwait()\n");
 424             stat = pthread_cond_timedwait(&jq->work, &jq->mutex, &timeout);
 425             if (stat == ETIMEDOUT) {
 426                Dmsg0(2300, "timedwait timedout.\n");
 427                timedout = true;
 428                break;
 429             } else if (stat != 0) {
 430                /* This shouldn't happen */
 431                Dmsg0(2300, "This shouldn't happen\n");
 432                jq->num_workers--;
 433                pthread_mutex_unlock(&jq->mutex);
 434                return NULL;
 435             }
 436             break;
 437          }
 438       }
 439       /*
 440        * If anything is in the ready queue, run it
 441        */
 442       Dmsg0(2300, "Checking ready queue.\n");
 443       while (!jq->ready_jobs->empty() && !jq->quit) {
 444          JCR *jcr;
 445          je = (jobq_item_t *)jq->ready_jobs->first();
 446          jcr = je->jcr;
 447          jq->ready_jobs->remove(je);
 448          if (!jq->ready_jobs->empty()) {
 449             Dmsg0(2300, "ready queue not empty start server\n");
 450             if (start_server(jq) != 0) {
 451                jq->num_workers--;
 452                pthread_mutex_unlock(&jq->mutex);
 453                return NULL;
 454             }
 455          }
 456          jq->running_jobs->append(je);
 457          set_jcr_in_tsd(jcr);
 458          Dmsg1(2300, "Took jobid=%d from ready and appended to run\n", jcr->JobId);
 459
 460          /* Release job queue lock */
 461          V(jq->mutex);
 462
 463          /* Call user's routine here */
 464          Dmsg2(2300, "Calling user engine for jobid=%d use=%d\n", jcr->JobId,
 465             jcr->use_count());
 466          jq->engine(je->jcr);
 467
 468          Dmsg2(2300, "Back from user engine jobid=%d use=%d.\n", jcr->JobId,
 469             jcr->use_count());
 470
 471          /* Reacquire job queue lock */
 472          P(jq->mutex);
 473          Dmsg0(200, "Done lock mutex after running job. Release locks.\n");
 474          jq->running_jobs->remove(je);
 475          /*
 476           * Release locks if acquired. Note, they will not have
 477           *  been acquired for jobs canceled before they were
 478           *  put into the ready queue.
 479           */
 480          if (jcr->acquired_resource_locks) {
 481             dec_read_store(jcr);
 482             dec_write_store(jcr);
 483             jcr->client->NumConcurrentJobs--;
 484             jcr->job->NumConcurrentJobs--;
 485             jcr->acquired_resource_locks = false;
 486          }
 487
 488          if (reschedule_job(jcr, jq, je)) {
 489             continue;              /* go look for more work */
 490          }
 491
 492          /* Clean up and release old jcr */
 493          Dmsg2(2300, "====== Termination job=%d use_cnt=%d\n", jcr->JobId, jcr->use_count());
 494          jcr->SDJobStatus = 0;
 495          V(jq->mutex);                /* release internal lock */
 496          free_jcr(jcr);
 497          free(je);                    /* release job entry */
 498          P(jq->mutex);                /* reacquire job queue lock */
 499       }
 500       /*
 501        * If any job in the wait queue can be run,
 502        *  move it to the ready queue
 503        */
 504       Dmsg0(2300, "Done check ready, now check wait queue.\n");
 505       if (!jq->waiting_jobs->empty() && !jq->quit) {
 506          int Priority;
 507          je = (jobq_item_t *)jq->waiting_jobs->first();
 508          jobq_item_t *re = (jobq_item_t *)jq->running_jobs->first();
 509          if (re) {
 510             Priority = re->jcr->JobPriority;
 511             Dmsg2(2300, "JobId %d is running. Look for pri=%d\n", re->jcr->JobId, Priority);
 512          } else {
 513             Priority = je->jcr->JobPriority;
 514             Dmsg1(2300, "No job running. Look for Job pri=%d\n", Priority);
 515          }
 516          /*
 517           * Walk down the list of waiting jobs and attempt
 518           *   to acquire the resources it needs.
 519           */
 520          for ( ; je;  ) {
 521             /* je is current job item on the queue, jn is the next one */
 522             JCR *jcr = je->jcr;
 523             jobq_item_t *jn = (jobq_item_t *)jq->waiting_jobs->next(je);
 524
 525             Dmsg3(2300, "Examining Job=%d JobPri=%d want Pri=%d\n",
 526                jcr->JobId, jcr->JobPriority, Priority);
 527
 528             /* Take only jobs of correct Priority */
 529             if (jcr->JobPriority != Priority) {
 530                set_jcr_job_status(jcr, JS_WaitPriority);
 531                break;
 532             }
 533
 534             if (!acquire_resources(jcr)) {
 535                /* If resource conflict, job is canceled */
 536                if (!job_canceled(jcr)) {
 537                   je = jn;            /* point to next waiting job */
 538                   continue;
 539                }
 540             }
 541
 542             /*
 543              * Got all locks, now remove it from wait queue and append it
 544              *   to the ready queue.  Note, we may also get here if the
 545              *    job was canceled.  Once it is "run", it will quickly
 546              *    terminate.
 547              */
 548             jq->waiting_jobs->remove(je);
 549             jq->ready_jobs->append(je);
 550             Dmsg1(2300, "moved JobId=%d from wait to ready queue\n", je->jcr->JobId);
 551             je = jn;                  /* Point to next waiting job */
 552          } /* end for loop */
 553
 554       } /* end if */
 555
 556       Dmsg0(2300, "Done checking wait queue.\n");
 557       /*
 558        * If no more ready work and we are asked to quit, then do it
 559        */
 560       if (jq->ready_jobs->empty() && jq->quit) {
 561          jq->num_workers--;
 562          if (jq->num_workers == 0) {
 563             Dmsg0(2300, "Wake up destroy routine\n");
 564             /* Wake up destroy routine if he is waiting */
 565             pthread_cond_broadcast(&jq->work);
 566          }
 567          break;
 568       }
 569       Dmsg0(2300, "Check for work request\n");
 570       /*
 571        * If no more work requests, and we waited long enough, quit
 572        */
 573       Dmsg2(2300, "timedout=%d read empty=%d\n", timedout,
 574          jq->ready_jobs->empty());
 575       if (jq->ready_jobs->empty() && timedout) {
 576          Dmsg0(2300, "break big loop\n");
 577          jq->num_workers--;
 578          break;
 579       }
 580
 581       work = !jq->ready_jobs->empty() || !jq->waiting_jobs->empty();
 582       if (work) {
 583          /*
 584           * If a job is waiting on a Resource, don't consume all
 585           *   the CPU time looping looking for work, and even more
 586           *   important, release the lock so that a job that has
 587           *   terminated can give us the resource.
 588           */
 589          V(jq->mutex);
 590          bmicrosleep(2, 0);              /* pause for 2 seconds */
 591          P(jq->mutex);
 592          /* Recompute work as something may have changed in last 2 secs */
 593          work = !jq->ready_jobs->empty() || !jq->waiting_jobs->empty();
 594       }
 595       Dmsg1(2300, "Loop again. work=%d\n", work);
 596    } /* end of big for loop */
 597
 598    Dmsg0(200, "unlock mutex\n");
 599    V(jq->mutex);
 600    Dmsg0(2300, "End jobq_server\n");
 601    return NULL;
 602 }
 603
 604 /*
 605  * Returns true if cleanup done and we should look for more work
 606  */
 607 static bool reschedule_job(JCR *jcr, jobq_t *jq, jobq_item_t *je)
 608 {
 609    /*
 610     * Reschedule the job if necessary and requested
 611     */
 612    if (jcr->job->RescheduleOnError &&
 613        jcr->JobStatus != JS_Terminated &&
 614        jcr->JobStatus != JS_Canceled &&
 615        jcr->get_JobType() == JT_BACKUP &&
 616        (jcr->job->RescheduleTimes == 0 ||
 617         jcr->reschedule_count < jcr->job->RescheduleTimes)) {
 618        char dt[50], dt2[50];
 619
 620        /*
 621         * Reschedule this job by cleaning it up, but
 622         *  reuse the same JobId if possible.
 623         */
 624       time_t now = time(NULL);
 625       jcr->reschedule_count++;
 626       jcr->sched_time = now + jcr->job->RescheduleInterval;
 627       bstrftime(dt, sizeof(dt), now);
 628       bstrftime(dt2, sizeof(dt2), jcr->sched_time);
 629       Dmsg4(2300, "Rescheduled Job %s to re-run in %d seconds.(now=%u,then=%u)\n", jcr->Job,
 630             (int)jcr->job->RescheduleInterval, now, jcr->sched_time);
 631       Jmsg(jcr, M_INFO, 0, _("Rescheduled Job %s at %s to re-run in %d seconds (%s).\n"),
 632            jcr->Job, dt, (int)jcr->job->RescheduleInterval, dt2);
 633       dird_free_jcr_pointers(jcr);     /* partial cleanup old stuff */
 634       jcr->JobStatus = -1;
 635       set_jcr_job_status(jcr, JS_WaitStartTime);
 636       jcr->SDJobStatus = 0;
 637       if (!allow_duplicate_job(jcr)) {
 638          return false;
 639       }
 640       if (jcr->JobBytes == 0) {
 641          Dmsg2(2300, "Requeue job=%d use=%d\n", jcr->JobId, jcr->use_count());
 642          V(jq->mutex);
 643          jobq_add(jq, jcr);     /* queue the job to run again */
 644          P(jq->mutex);
 645          free_jcr(jcr);         /* release jcr */
 646          free(je);              /* free the job entry */
 647          return true;           /* we already cleaned up */
 648       }
 649       /*
 650        * Something was actually backed up, so we cannot reuse
 651        *   the old JobId or there will be database record
 652        *   conflicts.  We now create a new job, copying the
 653        *   appropriate fields.
 654        */
 655       JCR *njcr = new_jcr(sizeof(JCR), dird_free_jcr);
 656       set_jcr_defaults(njcr, jcr->job);
 657       njcr->reschedule_count = jcr->reschedule_count;
 658       njcr->sched_time = jcr->sched_time;
 659       njcr->set_JobLevel(jcr->get_JobLevel());
 660       njcr->pool = jcr->pool;
 661       njcr->run_pool_override = jcr->run_pool_override;
 662       njcr->full_pool = jcr->full_pool;
 663       njcr->run_full_pool_override = jcr->run_full_pool_override;
 664       njcr->inc_pool = jcr->inc_pool;
 665       njcr->run_inc_pool_override = jcr->run_inc_pool_override;
 666       njcr->diff_pool = jcr->diff_pool;
 667       njcr->JobStatus = -1;
 668       set_jcr_job_status(njcr, jcr->JobStatus);
 669       if (jcr->rstore) {
 670          copy_rstorage(njcr, jcr->rstorage, _("previous Job"));
 671       } else {
 672          free_rstorage(njcr);
 673       }
 674       if (jcr->wstore) {
 675          copy_wstorage(njcr, jcr->wstorage, _("previous Job"));
 676       } else {
 677          free_wstorage(njcr);
 678       }
 679       njcr->messages = jcr->messages;
 680       njcr->spool_data = jcr->spool_data;
 681       njcr->write_part_after_job = jcr->write_part_after_job;
 682       Dmsg0(2300, "Call to run new job\n");
 683       V(jq->mutex);
 684       run_job(njcr);            /* This creates a "new" job */
 685       free_jcr(njcr);           /* release "new" jcr */
 686       P(jq->mutex);
 687       Dmsg0(2300, "Back from running new job.\n");
 688    }
 689    return false;
 690 }
 691
 692 /*
 693  * See if we can acquire all the necessary resources for the job (JCR)
 694  *
 695  *  Returns: true  if successful
 696  *           false if resource failure
 697  */
 698 static bool acquire_resources(JCR *jcr)
 699 {
 700    bool skip_this_jcr = false;
 701
 702    jcr->acquired_resource_locks = false;
 703    if (jcr->rstore && jcr->rstore == jcr->wstore) {    /* possible deadlock */
 704       Jmsg(jcr, M_FATAL, 0, _("Job canceled. Attempt to read and write same device.\n"
 705          "    Read storage \"%s\" (From %s) -- Write storage \"%s\" (From %s)\n"),
 706          jcr->rstore->name(), jcr->rstore_source, jcr->wstore->name(), jcr->wstore_source);
 707       set_jcr_job_status(jcr, JS_Canceled);
 708       return false;
 709    }
 710    if (jcr->rstore) {
 711       Dmsg1(200, "Rstore=%s\n", jcr->rstore->name());
 712       if (jcr->rstore->NumConcurrentJobs < jcr->rstore->MaxConcurrentJobs) {
 713          jcr->rstore->NumConcurrentReadJobs++;
 714          jcr->rstore->NumConcurrentJobs++;
 715          Dmsg1(200, "Inc rncj=%d\n", jcr->rstore->NumConcurrentJobs);
 716       } else {
 717          Dmsg1(200, "Fail rncj=%d\n", jcr->rstore->NumConcurrentJobs);
 718          set_jcr_job_status(jcr, JS_WaitStoreRes);
 719          return false;
 720       }
 721    }
 722
 723    if (jcr->wstore) {
 724       Dmsg1(200, "Wstore=%s\n", jcr->wstore->name());
 725       if (jcr->wstore->NumConcurrentJobs < jcr->wstore->MaxConcurrentJobs) {
 726          jcr->wstore->NumConcurrentJobs++;
 727          Dmsg1(200, "Inc wncj=%d\n", jcr->wstore->NumConcurrentJobs);
 728       } else if (jcr->rstore) {
 729          dec_read_store(jcr);
 730          skip_this_jcr = true;
 731       } else {
 732          Dmsg1(200, "Fail wncj=%d\n", jcr->wstore->NumConcurrentJobs);
 733          skip_this_jcr = true;
 734       }
 735    }
 736    if (skip_this_jcr) {
 737       set_jcr_job_status(jcr, JS_WaitStoreRes);
 738       return false;
 739    }
 740
 741    if (jcr->client->NumConcurrentJobs < jcr->client->MaxConcurrentJobs) {
 742       jcr->client->NumConcurrentJobs++;
 743    } else {
 744       /* Back out previous locks */
 745       dec_write_store(jcr);
 746       dec_read_store(jcr);
 747       set_jcr_job_status(jcr, JS_WaitClientRes);
 748       return false;
 749    }
 750    if (jcr->job->NumConcurrentJobs < jcr->job->MaxConcurrentJobs) {
 751       jcr->job->NumConcurrentJobs++;
 752    } else {
 753       /* Back out previous locks */
 754       dec_write_store(jcr);
 755       dec_read_store(jcr);
 756       jcr->client->NumConcurrentJobs--;
 757       set_jcr_job_status(jcr, JS_WaitJobRes);
 758       return false;
 759    }
 760
 761    jcr->acquired_resource_locks = true;
 762    return true;
 763 }
 764
 765 static void dec_read_store(JCR *jcr)
 766 {
 767    if (jcr->rstore) {
 768       jcr->rstore->NumConcurrentReadJobs--;    /* back out rstore */
 769       jcr->rstore->NumConcurrentJobs--;        /* back out rstore */
 770       Dmsg1(200, "Dec rncj=%d\n", jcr->rstore->NumConcurrentJobs);
 771       ASSERT(jcr->rstore->NumConcurrentReadJobs >= 0);
 772       ASSERT(jcr->rstore->NumConcurrentJobs >= 0);
 773    }
 774 }
 775
 776 static void dec_write_store(JCR *jcr)
 777 {
 778    if (jcr->wstore) {
 779       jcr->wstore->NumConcurrentJobs--;
 780       Dmsg1(200, "Dec wncj=%d\n", jcr->wstore->NumConcurrentJobs);
 781       ASSERT(jcr->wstore->NumConcurrentJobs >= 0);
 782    }
 783 }