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) 2000-2003 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 #ifdef JOB_QUEUE
  44
  45 /* Forward referenced functions */
  46 static void *jobq_server(void *arg);
  47 static int   start_server(jobq_t *jq);
  48
  49 /*
  50  * Initialize a job queue
  51  *
  52  *  Returns: 0 on success
  53  *           errno on failure
  54  */
  55 int jobq_init(jobq_t *jq, int threads, void *(*engine)(void *arg))
  56 {
  57    int stat;
  58    jobq_item_t *item = NULL;
  59
  60    if ((stat = pthread_attr_init(&jq->attr)) != 0) {
  61       return stat;
  62    }
  63    if ((stat = pthread_attr_setdetachstate(&jq->attr, PTHREAD_CREATE_DETACHED)) != 0) {
  64       pthread_attr_destroy(&jq->attr);
  65       return stat;
  66    }
  67    if ((stat = pthread_mutex_init(&jq->mutex, NULL)) != 0) {
  68       pthread_attr_destroy(&jq->attr);
  69       return stat;
  70    }
  71    if ((stat = pthread_cond_init(&jq->work, NULL)) != 0) {
  72       pthread_mutex_destroy(&jq->mutex);
  73       pthread_attr_destroy(&jq->attr);
  74       return stat;
  75    }
  76    jq->quit = false;
  77    jq->max_workers = threads;         /* max threads to create */
  78    jq->num_workers = 0;               /* no threads yet */
  79    jq->idle_workers = 0;              /* no idle threads */
  80    jq->engine = engine;               /* routine to run */
  81    jq->valid = JOBQ_VALID;
  82    /* Initialize the job queues */
  83    jq->waiting_jobs = new dlist(item, &item->link);
  84    jq->running_jobs = new dlist(item, &item->link);
  85    jq->ready_jobs = new dlist(item, &item->link);
  86    return 0;
  87 }
  88
  89 /*
  90  * Destroy the job queue
  91  *
  92  * Returns: 0 on success
  93  *          errno on failure
  94  */
  95 int jobq_destroy(jobq_t *jq)
  96 {
  97    int stat, stat1, stat2;
  98
  99   if (jq->valid != JOBQ_VALID) {
 100      return EINVAL;
 101   }
 102   if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 103      return stat;
 104   }
 105   jq->valid = 0;                      /* prevent any more operations */
 106
 107   /*
 108    * If any threads are active, wake them
 109    */
 110   if (jq->num_workers > 0) {
 111      jq->quit = true;
 112      if (jq->idle_workers) {
 113         if ((stat = pthread_cond_broadcast(&jq->work)) != 0) {
 114            pthread_mutex_unlock(&jq->mutex);
 115            return stat;
 116         }
 117      }
 118      while (jq->num_workers > 0) {
 119         if ((stat = pthread_cond_wait(&jq->work, &jq->mutex)) != 0) {
 120            pthread_mutex_unlock(&jq->mutex);
 121            return stat;
 122         }
 123      }
 124   }
 125   if ((stat = pthread_mutex_unlock(&jq->mutex)) != 0) {
 126      return stat;
 127   }
 128   stat  = pthread_mutex_destroy(&jq->mutex);
 129   stat1 = pthread_cond_destroy(&jq->work);
 130   stat2 = pthread_attr_destroy(&jq->attr);
 131   delete jq->waiting_jobs;
 132   delete jq->running_jobs;
 133   delete jq->ready_jobs;
 134   return (stat != 0 ? stat : (stat1 != 0 ? stat1 : stat2));
 135 }
 136
 137 struct wait_pkt {
 138    JCR *jcr;
 139    jobq_t *jq;
 140 };
 141
 142 /*
 143  * Wait until schedule time arrives before starting. Normally
 144  *  this routine is only used for jobs started from the console
 145  *  for which the user explicitly specified a start time. Otherwise
 146  *  most jobs are put into the job queue only when their
 147  *  scheduled time arives.
 148  */
 149 static void *sched_wait(void *arg)
 150 {
 151    JCR *jcr = ((wait_pkt *)arg)->jcr;
 152    jobq_t *jq = ((wait_pkt *)arg)->jq;
 153
 154    Dmsg0(100, "Enter sched_wait.\n");
 155    free(arg);
 156    time_t wtime = jcr->sched_time - time(NULL);
 157    /* Wait until scheduled time arrives */
 158    if (wtime > 0 && verbose) {
 159       Jmsg(jcr, M_INFO, 0, _("Job %s waiting %d seconds for scheduled start time.\n"),
 160          jcr->Job, wtime);
 161       set_jcr_job_status(jcr, JS_WaitStartTime);
 162    }
 163    /* Check every 30 seconds if canceled */
 164    while (wtime > 0) {
 165       Dmsg2(100, "Waiting on sched time, jobid=%d secs=%d\n", jcr->JobId, wtime);
 166       if (wtime > 30) {
 167          wtime = 30;
 168       }
 169       bmicrosleep(wtime, 0);
 170       if (job_canceled(jcr)) {
 171          break;
 172       }
 173       wtime = jcr->sched_time - time(NULL);
 174    }
 175    jobq_add(jq, jcr);
 176    Dmsg0(100, "Exit sched_wait\n");
 177    return NULL;
 178 }
 179
 180
 181 /*
 182  *  Add a job to the queue
 183  *    jq is a queue that was created with jobq_init
 184  *
 185  */
 186 int jobq_add(jobq_t *jq, JCR *jcr)
 187 {
 188    int stat;
 189    jobq_item_t *item, *li;
 190    bool inserted = false;
 191    time_t wtime = jcr->sched_time - time(NULL);
 192    pthread_t id;
 193    wait_pkt *sched_pkt;
 194
 195
 196    Dmsg1(100, "jobq_add jobid=%d\n", jcr->JobId);
 197    if (jq->valid != JOBQ_VALID) {
 198       return EINVAL;
 199    }
 200
 201    if (!job_canceled(jcr) && wtime > 0) {
 202       set_thread_concurrency(jq->max_workers + 2);
 203       sched_pkt = (wait_pkt *)malloc(sizeof(wait_pkt));
 204       sched_pkt->jcr = jcr;
 205       sched_pkt->jq = jq;
 206       stat = pthread_create(&id, &jq->attr, sched_wait, (void *)sched_pkt);
 207       return stat;
 208    }
 209
 210    if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 211       return stat;
 212    }
 213
 214    if ((item = (jobq_item_t *)malloc(sizeof(jobq_item_t))) == NULL) {
 215       return ENOMEM;
 216    }
 217    item->jcr = jcr;
 218
 219    if (job_canceled(jcr)) {
 220       /* Add job to ready queue so that it is canceled quickly */
 221       jq->ready_jobs->prepend(item);
 222       Dmsg1(100, "Prepended job=%d to ready queue\n", jcr->JobId);
 223    } else {
 224       /* Add this job to the wait queue in priority sorted order */
 225       for (li=NULL; (li=(jobq_item_t *)jq->waiting_jobs->next(li)); ) {
 226          Dmsg2(100, "waiting item jobid=%d priority=%d\n",
 227             li->jcr->JobId, li->jcr->JobPriority);
 228          if (li->jcr->JobPriority > jcr->JobPriority) {
 229             jq->waiting_jobs->insert_before(item, li);
 230             Dmsg2(100, "insert_before jobid=%d before %d\n",
 231                li->jcr->JobId, jcr->JobId);
 232             inserted = true;
 233             break;
 234          }
 235       }
 236       /* If not jobs in wait queue, append it */
 237       if (!inserted) {
 238          jq->waiting_jobs->append(item);
 239          Dmsg1(100, "Appended item jobid=%d\n", jcr->JobId);
 240       }
 241    }
 242
 243    /* Ensure that at least one server looks at the queue. */
 244    stat = start_server(jq);
 245
 246    if (stat == 0) {
 247       pthread_mutex_unlock(&jq->mutex);
 248    }
 249    Dmsg0(100, "Return jobq_add\n");
 250    return stat;
 251 }
 252
 253 /*
 254  *  Remove a job from the job queue. Used only by cancel Console command.
 255  *    jq is a queue that was created with jobq_init
 256  *    work_item is an element of work
 257  *
 258  *   Note, it is "removed" by immediately calling a processing routine.
 259  *    if you want to cancel it, you need to provide some external means
 260  *    of doing so.
 261  */
 262 int jobq_remove(jobq_t *jq, JCR *jcr)
 263 {
 264    int stat;
 265    bool found = false;
 266    jobq_item_t *item;
 267
 268    Dmsg1(100, "jobq_remove jobid=%d\n", jcr->JobId);
 269    if (jq->valid != JOBQ_VALID) {
 270       return EINVAL;
 271    }
 272
 273    if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 274       return stat;
 275    }
 276
 277    for (item=NULL; (item=(jobq_item_t *)jq->waiting_jobs->next(item)); ) {
 278       if (jcr == item->jcr) {
 279          found = true;
 280          break;
 281       }
 282    }
 283    if (!found) {
 284       return EINVAL;
 285    }
 286
 287    /* Move item to be the first on the list */
 288    jq->waiting_jobs->remove(item);
 289    jq->ready_jobs->prepend(item);
 290
 291    stat = start_server(jq);
 292    if (stat != 0) {
 293       return stat;
 294    }
 295    pthread_mutex_unlock(&jq->mutex);
 296    Dmsg0(100, "Return jobq_remove\n");
 297    return stat;
 298 }
 299
 300
 301 /*
 302  * Start the server thread if it isn't already running
 303  */
 304 static int start_server(jobq_t *jq)
 305 {
 306    int stat = 0;
 307    pthread_t id;
 308
 309    /* if any threads are idle, wake one */
 310    if (jq->idle_workers > 0) {
 311       Dmsg0(100, "Signal worker to wake up\n");
 312       if ((stat = pthread_cond_signal(&jq->work)) != 0) {
 313          pthread_mutex_unlock(&jq->mutex);
 314          return stat;
 315       }
 316    } else if (jq->num_workers < jq->max_workers) {
 317       Dmsg0(100, "Create worker thread\n");
 318       /* No idle threads so create a new one */
 319       set_thread_concurrency(jq->max_workers + 1);
 320       if ((stat = pthread_create(&id, &jq->attr, jobq_server, (void *)jq)) != 0) {
 321          pthread_mutex_unlock(&jq->mutex);
 322          return stat;
 323       }
 324       jq->num_workers++;
 325    }
 326    return stat;
 327 }
 328
 329
 330 /*
 331  * This is the worker thread that serves the job queue.
 332  * When all the resources are acquired for the job,
 333  *  it will call the user's engine.
 334  */
 335 static void *jobq_server(void *arg)
 336 {
 337    struct timespec timeout;
 338    jobq_t *jq = (jobq_t *)arg;
 339    jobq_item_t *je;                   /* job entry in queue */
 340    int stat;
 341    bool timedout;
 342    bool work = true;
 343
 344    Dmsg0(100, "Start jobq_server\n");
 345    if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 346       return NULL;
 347    }
 348
 349    for (;;) {
 350       struct timeval tv;
 351       struct timezone tz;
 352
 353       Dmsg0(100, "Top of for loop\n");
 354       timedout = false;
 355       Dmsg0(100, "gettimeofday()\n");
 356       gettimeofday(&tv, &tz);
 357       timeout.tv_nsec = 0;
 358       timeout.tv_sec = tv.tv_sec + 4;
 359
 360       while (!work && !jq->quit) {
 361          /*
 362           * Wait 4 seconds, then if no more work, exit
 363           */
 364          Dmsg0(200, "pthread_cond_timedwait()\n");
 365          stat = pthread_cond_timedwait(&jq->work, &jq->mutex, &timeout);
 366          Dmsg1(100, "timedwait=%d\n", stat);
 367          if (stat == ETIMEDOUT) {
 368             timedout = true;
 369             break;
 370          } else if (stat != 0) {
 371             /* This shouldn't happen */
 372             Dmsg0(100, "This shouldn't happen\n");
 373             jq->num_workers--;
 374             pthread_mutex_unlock(&jq->mutex);
 375             return NULL;
 376          }
 377       }
 378       /*
 379        * If anything is in the ready queue, run it
 380        */
 381       Dmsg0(100, "Checking ready queue.\n");
 382       while (!jq->ready_jobs->empty() && !jq->quit) {
 383          JCR *jcr;
 384          je = (jobq_item_t *)jq->ready_jobs->first();
 385          jcr = je->jcr;
 386          jq->ready_jobs->remove(je);
 387          if (!jq->ready_jobs->empty()) {
 388             Dmsg0(100, "ready queue not empty start server\n");
 389             if (start_server(jq) != 0) {
 390                return NULL;
 391             }
 392          }
 393          jq->running_jobs->append(je);
 394          Dmsg1(100, "Took jobid=%d from ready and appended to run\n", jcr->JobId);
 395          if ((stat = pthread_mutex_unlock(&jq->mutex)) != 0) {
 396             return NULL;
 397          }
 398          /* Call user's routine here */
 399          Dmsg1(100, "Calling user engine for jobid=%d\n", jcr->JobId);
 400          jq->engine(je->jcr);
 401          Dmsg1(100, "Back from user engine jobid=%d.\n", jcr->JobId);
 402          if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
 403             free(je);                 /* release job entry */
 404             return NULL;
 405          }
 406          Dmsg0(200, "Done lock mutex\n");
 407          jq->running_jobs->remove(je);
 408          /*
 409           * Release locks if acquired. Note, they will not have
 410           *  been acquired for jobs canceled before they were
 411           *  put into the ready queue.
 412           */
 413          if (jcr->acquired_resource_locks) {
 414             jcr->store->NumConcurrentJobs--;
 415             jcr->client->NumConcurrentJobs--;
 416             jcr->job->NumConcurrentJobs--;
 417          }
 418
 419          if (jcr->job->RescheduleOnError &&
 420              jcr->JobStatus != JS_Terminated &&
 421              jcr->JobStatus != JS_Canceled &&
 422              jcr->job->RescheduleTimes > 0 &&
 423              jcr->reschedule_count < jcr->job->RescheduleTimes) {
 424
 425              /*
 426               * Reschedule this job by cleaning it up, but
 427               *  reuse the same JobId if possible.
 428               */
 429             jcr->reschedule_count++;
 430             jcr->sched_time = time(NULL) + jcr->job->RescheduleInterval;
 431             Dmsg2(100, "Rescheduled Job %s to re-run in %d seconds.\n", jcr->Job,
 432                (int)jcr->job->RescheduleInterval);
 433             jcr->JobStatus = JS_Created; /* force new status */
 434             dird_free_jcr(jcr);          /* partial cleanup old stuff */
 435             if (jcr->JobBytes == 0) {
 436                Dmsg1(100, "Requeue job=%d\n", jcr->JobId);
 437                V(jq->mutex);
 438                jobq_add(jq, jcr);     /* queue the job to run again */
 439                P(jq->mutex);
 440                free(je);              /* free the job entry */
 441                continue;              /* look for another job to run */
 442             }
 443             /*
 444              * Something was actually backed up, so we cannot reuse
 445              *   the old JobId or there will be database record
 446              *   conflicts.  We now create a new job, copying the
 447              *   appropriate fields.
 448              */
 449             JCR *njcr = new_jcr(sizeof(JCR), dird_free_jcr);
 450             set_jcr_defaults(njcr, jcr->job);
 451             njcr->reschedule_count = jcr->reschedule_count;
 452             njcr->JobLevel = jcr->JobLevel;
 453             njcr->JobStatus = jcr->JobStatus;
 454             njcr->pool = jcr->pool;
 455             njcr->store = jcr->store;
 456             njcr->messages = jcr->messages;
 457             Dmsg0(100, "Call to run new job\n");
 458             V(jq->mutex);
 459             run_job(njcr);            /* This creates a "new" job */
 460             P(jq->mutex);
 461             Dmsg0(100, "Back from running new job.\n");
 462          }
 463          /* Clean up and release old jcr */
 464          if (jcr->db) {
 465             Dmsg0(100, "Close DB\n");
 466             db_close_database(jcr, jcr->db);
 467             jcr->db = NULL;
 468          }
 469          Dmsg1(100, "====== Termination job=%d\n", jcr->JobId);
 470          free_jcr(jcr);
 471          free(je);                    /* release job entry */
 472       }
 473       /*
 474        * If any job in the wait queue can be run,
 475        *  move it to the ready queue
 476        */
 477       Dmsg0(100, "Done check ready, now check wait queue.\n");
 478       while (!jq->waiting_jobs->empty() && !jq->quit) {
 479          int Priority;
 480          je = (jobq_item_t *)jq->waiting_jobs->first();
 481          jobq_item_t *re = (jobq_item_t *)jq->running_jobs->first();
 482          if (re) {
 483             Priority = re->jcr->JobPriority;
 484             Dmsg1(100, "Set Run pri=%d\n", Priority);
 485          } else {
 486             Priority = je->jcr->JobPriority;
 487             Dmsg1(100, "Set Job pri=%d\n", Priority);
 488          }
 489          /*
 490           * Walk down the list of waiting jobs and attempt
 491           *   to acquire the resources it needs.
 492           */
 493          for ( ; je;  ) {
 494             /* je is current job item on the queue, jn is the next one */
 495             JCR *jcr = je->jcr;
 496             jobq_item_t *jn = (jobq_item_t *)jq->waiting_jobs->next(je);
 497             Dmsg3(100, "Examining Job=%d JobPri=%d want Pri=%d\n",
 498                jcr->JobId, jcr->JobPriority, Priority);
 499             /* Take only jobs of correct Priority */
 500             if (jcr->JobPriority != Priority) {
 501                set_jcr_job_status(jcr, JS_WaitPriority);
 502                break;
 503             }
 504             if (jcr->store->NumConcurrentJobs < jcr->store->MaxConcurrentJobs) {
 505                jcr->store->NumConcurrentJobs++;
 506             } else {
 507                set_jcr_job_status(jcr, JS_WaitStoreRes);
 508                je = jn;
 509                continue;
 510             }
 511             if (jcr->client->NumConcurrentJobs < jcr->client->MaxConcurrentJobs) {
 512                jcr->client->NumConcurrentJobs++;
 513             } else {
 514                jcr->store->NumConcurrentJobs--;
 515                set_jcr_job_status(jcr, JS_WaitClientRes);
 516                je = jn;
 517                continue;
 518             }
 519             if (jcr->job->NumConcurrentJobs < jcr->job->MaxConcurrentJobs) {
 520                jcr->job->NumConcurrentJobs++;
 521             } else {
 522                jcr->store->NumConcurrentJobs--;
 523                jcr->client->NumConcurrentJobs--;
 524                set_jcr_job_status(jcr, JS_WaitJobRes);
 525                je = jn;
 526                continue;
 527             }
 528             /* Got all locks, now remove it from wait queue and append it
 529              *   to the ready queue
 530              */
 531             jcr->acquired_resource_locks = true;
 532             jq->waiting_jobs->remove(je);
 533             jq->ready_jobs->append(je);
 534             Dmsg1(100, "moved JobId=%d from wait to ready queue\n", je->jcr->JobId);
 535             je = jn;
 536          } /* end for loop */
 537          break;
 538       } /* end while loop */
 539       Dmsg0(100, "Done checking wait queue.\n");
 540       /*
 541        * If no more ready work and we are asked to quit, then do it
 542        */
 543       if (jq->ready_jobs->empty() && jq->quit) {
 544          jq->num_workers--;
 545          if (jq->num_workers == 0) {
 546             Dmsg0(100, "Wake up destroy routine\n");
 547             /* Wake up destroy routine if he is waiting */
 548             pthread_cond_broadcast(&jq->work);
 549          }
 550          break;
 551       }
 552       Dmsg0(100, "Check for work request\n");
 553       /*
 554        * If no more work requests, and we waited long enough, quit
 555        */
 556       Dmsg2(100, "timedout=%d read empty=%d\n", timedout,
 557          jq->ready_jobs->empty());
 558       if (jq->ready_jobs->empty() && timedout) {
 559          Dmsg0(100, "break big loop\n");
 560          jq->num_workers--;
 561          break;
 562       }
 563       Dmsg0(100, "Loop again\n");
 564       work = false;
 565    } /* end of big for loop */
 566
 567    Dmsg0(200, "unlock mutex\n");
 568    pthread_mutex_unlock(&jq->mutex);
 569    Dmsg0(100, "End jobq_server\n");
 570    return NULL;
 571 }
 572
 573 #endif /* JOB_QUEUE */