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

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