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