]> git.sur5r.net Git - bacula/bacula/blob - bacula/src/dird/job.c
139e88f7140e2152e036e92fd336a87c65ad6833
[bacula/bacula] / bacula / src / dird / job.c
1 /*
2  *
3  *   Bacula Director Job processing routines
4  *
5  *     Kern Sibbald, October MM
6  *
7  *    Version $Id$
8  */
9 /*
10    Copyright (C) 2000-2003 Kern Sibbald and John Walker
11
12    This program is free software; you can redistribute it and/or
13    modify it under the terms of the GNU General Public License as
14    published by the Free Software Foundation; either version 2 of
15    the License, or (at your option) any later version.
16
17    This program is distributed in the hope that it will be useful,
18    but WITHOUT ANY WARRANTY; without even the implied warranty of
19    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
20    General Public License for more details.
21
22    You should have received a copy of the GNU General Public
23    License along with this program; if not, write to the Free
24    Software Foundation, Inc., 59 Temple Place - Suite 330, Boston,
25    MA 02111-1307, USA.
26
27  */
28
29 #include "bacula.h"
30 #include "dird.h"
31
32 /* Forward referenced subroutines */
33 static void *job_thread(void *arg);
34 static char *edit_run_codes(JCR *jcr, char *omsg, char *imsg);
35 static void release_resource_locks(JCR *jcr);
36 static int acquire_resource_locks(JCR *jcr);
37 #ifdef USE_SEMAPHORE
38 static void backoff_resource_locks(JCR *jcr, int count);
39 #endif
40
41 /* Exported subroutines */
42 void run_job(JCR *jcr);
43
44
45 /* Imported subroutines */
46 extern void term_scheduler();
47 extern void term_ua_server();
48 extern int do_backup(JCR *jcr);
49 extern int do_restore(JCR *jcr);
50 extern int do_verify(JCR *jcr);
51 extern void backup_cleanup(void);
52
53 #ifdef USE_SEMAPHORE
54 static semlock_t job_lock;
55 static pthread_mutex_t mutex;
56 static pthread_cond_t  resource_wait;
57 static int waiting = 0;               /* count of waiting threads */
58 #else
59 /* Queue of jobs to be run */
60 workq_t job_wq;                   /* our job work queue */
61 #endif
62
63 void init_job_server(int max_workers)
64 {
65    int stat;
66 #ifdef USE_SEMAPHORE
67    if ((stat = sem_init(&job_lock, max_workers)) != 0) {
68       Emsg1(M_ABORT, 0, _("Could not init job lock: ERR=%s\n"), strerror(stat));
69    }
70    if ((stat = pthread_mutex_init(&mutex, NULL)) != 0) {
71       Emsg1(M_ABORT, 0, _("Could not init resource mutex: ERR=%s\n"), strerror(stat));
72    }
73    if ((stat = pthread_cond_init(&resource_wait, NULL)) != 0) {
74       Emsg1(M_ABORT, 0, _("Could not init resource wait: ERR=%s\n"), strerror(stat));
75    }
76
77 #else
78    if ((stat = workq_init(&job_wq, max_workers, job_thread)) != 0) {
79       Emsg1(M_ABORT, 0, _("Could not init job work queue: ERR=%s\n"), strerror(stat));
80    }
81 #endif
82    return;
83 }
84
85 /*
86  * Run a job -- typically called by the scheduler, but may also
87  *              be called by the UA (Console program).
88  *
89  */
90 void run_job(JCR *jcr)
91 {
92    int stat, errstat;
93 #ifdef USE_SEMAPHORE
94    pthread_t tid;
95 #else
96    workq_ele_t *work_item;
97 #endif
98
99    sm_check(__FILE__, __LINE__, True);
100    init_msg(jcr, jcr->messages);
101    create_unique_job_name(jcr, jcr->job->hdr.name);
102    set_jcr_job_status(jcr, JS_Created);
103    jcr->jr.SchedTime = jcr->sched_time;
104    jcr->jr.StartTime = jcr->start_time;
105    jcr->jr.Type = jcr->JobType;
106    jcr->jr.Level = jcr->JobLevel;
107    jcr->jr.JobStatus = jcr->JobStatus;
108    bstrncpy(jcr->jr.Name, jcr->job->hdr.name, sizeof(jcr->jr.Name));
109    bstrncpy(jcr->jr.Job, jcr->Job, sizeof(jcr->jr.Job));
110
111    /* Initialize termination condition variable */
112    if ((errstat = pthread_cond_init(&jcr->term_wait, NULL)) != 0) {
113       Jmsg1(jcr, M_FATAL, 0, _("Unable to init job cond variable: ERR=%s\n"), strerror(errstat));
114       set_jcr_job_status(jcr, JS_ErrorTerminated);
115       free_jcr(jcr);
116       return;
117    }
118
119    /*
120     * Open database
121     */
122    Dmsg0(50, "Open database\n");
123    jcr->db=db_init_database(jcr, jcr->catalog->db_name, jcr->catalog->db_user,
124                             jcr->catalog->db_password, jcr->catalog->db_address,
125                             jcr->catalog->db_port, jcr->catalog->db_socket);
126    if (!db_open_database(jcr, jcr->db)) {
127       Jmsg(jcr, M_FATAL, 0, "%s", db_strerror(jcr->db));
128       set_jcr_job_status(jcr, JS_ErrorTerminated);
129       free_jcr(jcr);
130       return;
131    }
132    Dmsg0(50, "DB opened\n");
133
134    /*
135     * Create Job record  
136     */
137    jcr->jr.JobStatus = jcr->JobStatus;
138    if (!db_create_job_record(jcr, jcr->db, &jcr->jr)) {
139       Jmsg(jcr, M_FATAL, 0, "%s", db_strerror(jcr->db));
140       set_jcr_job_status(jcr, JS_ErrorTerminated);
141       free_jcr(jcr);
142       return;
143    }
144    jcr->JobId = jcr->jr.JobId;
145    ASSERT(jcr->jr.JobId > 0);
146
147    Dmsg4(30, "Created job record JobId=%d Name=%s Type=%c Level=%c\n", 
148        jcr->JobId, jcr->Job, jcr->jr.Type, jcr->jr.Level);
149    Dmsg0(200, "Add jrc to work queue\n");
150
151 #ifdef USE_SEMAPHORE
152   if ((stat = pthread_create(&tid, NULL, job_thread, (void *)jcr)) != 0) {
153       Emsg1(M_ABORT, 0, _("Unable to create job thread: ERR=%s\n"), strerror(stat));
154    }
155 #else
156    /* Queue the job to be run */
157    if ((stat = workq_add(&job_wq, (void *)jcr, &work_item, 0)) != 0) {
158       Emsg1(M_ABORT, 0, _("Could not add job to work queue: ERR=%s\n"), strerror(stat));
159    }
160    jcr->work_item = work_item;
161 #endif
162    Dmsg0(200, "Done run_job()\n");
163 }
164
165 /* 
166  * This is the engine called by workq_add() when we were pulled                
167  *  from the work queue.
168  *  At this point, we are running in our own thread 
169  */
170 static void *job_thread(void *arg)
171 {
172    JCR *jcr = (JCR *)arg;
173
174    pthread_detach(pthread_self());
175    sm_check(__FILE__, __LINE__, True);
176
177    if (!acquire_resource_locks(jcr)) {
178       set_jcr_job_status(jcr, JS_Canceled);
179    }
180
181    Dmsg0(200, "=====Start Job=========\n");
182    jcr->start_time = time(NULL);      /* set the real start time */
183    set_jcr_job_status(jcr, JS_Running);
184
185    if (job_canceled(jcr)) {
186       update_job_end_record(jcr);
187    } else if (jcr->job->MaxStartDelay != 0 && jcr->job->MaxStartDelay <
188        (utime_t)(jcr->start_time - jcr->sched_time)) {
189       Jmsg(jcr, M_FATAL, 0, _("Job canceled because max start delay time exceeded.\n"));
190       set_jcr_job_status(jcr, JS_Canceled);
191       update_job_end_record(jcr);
192    } else {
193
194       /* Run Job */
195       if (jcr->job->RunBeforeJob) {
196          POOLMEM *before = get_pool_memory(PM_FNAME);
197          int status;
198          
199          before = edit_run_codes(jcr, before, jcr->job->RunBeforeJob);
200          status = run_program(before, 0, NULL);
201          if (status != 0) {
202             Jmsg(jcr, M_FATAL, 0, _("RunBeforeJob returned non-zero status=%d\n"),
203                status);
204             set_jcr_job_status(jcr, JS_FatalError);
205             update_job_end_record(jcr);
206             free_pool_memory(before);
207             goto bail_out;
208          }
209          free_pool_memory(before);
210       }
211       switch (jcr->JobType) {
212          case JT_BACKUP:
213             do_backup(jcr);
214             if (jcr->JobStatus == JS_Terminated) {
215                do_autoprune(jcr);
216             }
217             break;
218          case JT_VERIFY:
219             do_verify(jcr);
220             if (jcr->JobStatus == JS_Terminated) {
221                do_autoprune(jcr);
222             }
223             break;
224          case JT_RESTORE:
225             do_restore(jcr);
226             if (jcr->JobStatus == JS_Terminated) {
227                do_autoprune(jcr);
228             }
229             break;
230          case JT_ADMIN:
231             /* No actual job */
232             do_autoprune(jcr);
233             set_jcr_job_status(jcr, JS_Terminated);
234             break;
235          default:
236             Pmsg1(0, "Unimplemented job type: %d\n", jcr->JobType);
237             break;
238          }
239       if (jcr->job->RunAfterJob) {
240          POOLMEM *after = get_pool_memory(PM_FNAME);
241          int status;
242       
243          after = edit_run_codes(jcr, after, jcr->job->RunAfterJob);
244          status = run_program(after, 0, NULL);
245          if (status != 0) {
246             Jmsg(jcr, M_FATAL, 0, _("RunAfterJob returned non-zero status=%d\n"),
247                status);
248             set_jcr_job_status(jcr, JS_FatalError);
249             update_job_end_record(jcr);
250          }
251          free_pool_memory(after);
252       }
253    }
254 bail_out:
255    release_resource_locks(jcr);
256    Dmsg0(50, "Before free jcr\n");
257    free_jcr(jcr);
258    Dmsg0(50, "======== End Job ==========\n");
259    sm_check(__FILE__, __LINE__, True);
260    return NULL;
261 }
262
263 /*
264  * Acquire the resources needed. These locks limit the
265  *  number of jobs by each resource. We have limits on
266  *  Jobs, Clients, Storage, and total jobs.
267  */
268 static int acquire_resource_locks(JCR *jcr)
269 {
270    time_t now = time(NULL);
271
272    /* Wait until scheduled time arrives */
273    if (jcr->sched_time > now && verbose) {
274       Jmsg(jcr, M_INFO, 0, _("Waiting %d seconds for sched time.\n"), 
275            jcr->sched_time - now);
276    }
277    while (jcr->sched_time > now) {
278       Dmsg2(100, "Waiting on sched time, jobid=%d secs=%d\n", jcr->JobId,
279             jcr->sched_time - now);
280       bmicrosleep(jcr->sched_time - now, 0);
281       now = time(NULL);
282       if (job_canceled(jcr)) {
283          return 0;
284       }
285    }
286
287
288 #ifdef USE_SEMAPHORE
289    int stat;
290
291    /* Initialize semaphores */
292    if (jcr->store->sem.valid != SEMLOCK_VALID) {
293       if ((stat = sem_init(&jcr->store->sem, jcr->store->MaxConcurrentJobs)) != 0) {
294          Emsg1(M_ABORT, 0, _("Could not init Storage semaphore: ERR=%s\n"), strerror(stat));
295       }
296    }
297    if (jcr->client->sem.valid != SEMLOCK_VALID) {
298       if ((stat = sem_init(&jcr->client->sem, jcr->client->MaxConcurrentJobs)) != 0) {
299          Emsg1(M_ABORT, 0, _("Could not init Client semaphore: ERR=%s\n"), strerror(stat));
300       }
301    }
302    if (jcr->job->sem.valid != SEMLOCK_VALID) {
303       if ((stat = sem_init(&jcr->job->sem, jcr->job->MaxConcurrentJobs)) != 0) {
304          Emsg1(M_ABORT, 0, _("Could not init Job semaphore: ERR=%s\n"), strerror(stat));
305       }
306    }
307
308    for ( ;; ) {
309       /* Acquire semaphore */
310       set_jcr_job_status(jcr, JS_WaitJobRes);
311       if ((stat = sem_lock(&jcr->job->sem)) != 0) {
312          Emsg1(M_ABORT, 0, _("Could not acquire Job max jobs lock: ERR=%s\n"), strerror(stat));
313       }
314       set_jcr_job_status(jcr, JS_WaitClientRes);
315       if ((stat = sem_trylock(&jcr->client->sem)) != 0) {
316          if (stat == EBUSY) {
317             backoff_resource_locks(jcr, 1);
318             goto wait;
319          } else {
320             Emsg1(M_ABORT, 0, _("Could not acquire Client max jobs lock: ERR=%s\n"), strerror(stat));
321          }
322       }
323       set_jcr_job_status(jcr, JS_WaitStoreRes);
324       if ((stat = sem_trylock(&jcr->store->sem)) != 0) {
325          if (stat == EBUSY) {
326             backoff_resource_locks(jcr, 2);
327             goto wait;
328          } else {
329             Emsg1(M_ABORT, 0, _("Could not acquire Storage max jobs lock: ERR=%s\n"), strerror(stat));
330          }
331       }
332       set_jcr_job_status(jcr, JS_WaitMaxJobs);
333       if ((stat = sem_trylock(&job_lock)) != 0) {
334          if (stat == EBUSY) {
335             backoff_resource_locks(jcr, 3);
336             goto wait;
337          } else {
338             Emsg1(M_ABORT, 0, _("Could not acquire max jobs lock: ERR=%s\n"), strerror(stat));
339          }
340       }
341       break;
342
343 wait:
344       if (job_canceled(jcr)) {
345          return 0;
346       }
347       P(mutex);
348       /*
349        * Wait for a resource to be released either by backoff or
350        *  by a job terminating.
351        */
352       waiting++;
353       pthread_cond_wait(&resource_wait, &mutex);
354       waiting--;
355       V(mutex);
356       /* Try again */
357    }
358 #endif
359    return 1;
360 }
361
362 #ifdef USE_SEMAPHORE
363 /*
364  * We could not get all the resource locks because 
365  *  too many jobs are running, so release any locks
366  *  we did acquire, giving others a chance to use them
367  *  while we wait.
368  */
369 static void backoff_resource_locks(JCR *jcr, int count)
370 {
371    P(mutex);
372    switch (count) {
373    case 3:
374       sem_unlock(&jcr->store->sem);
375       /* Fall through wanted */
376    case 2:
377       sem_unlock(&jcr->client->sem);
378       /* Fall through wanted */
379    case 1:
380       sem_unlock(&jcr->job->sem);
381       break;
382    }
383    /*
384     * Since we released a lock, if there are any threads
385     *  waiting, wake them up so that they can try again.
386     */
387    if (waiting > 0) {
388       pthread_cond_broadcast(&resource_wait);
389    }
390    V(mutex);
391 }
392 #endif
393
394 /*
395  * This is called at the end of the job to release
396  *   any resource limits on the number of jobs. If
397  *   there are any other jobs waiting, we wake them
398  *   up so that they can try again.
399  */
400 static void release_resource_locks(JCR *jcr)
401 {
402 #ifdef USE_SEMAPHORE
403    P(mutex);
404    sem_unlock(&jcr->store->sem);
405    sem_unlock(&jcr->client->sem);
406    sem_unlock(&jcr->job->sem);
407    sem_unlock(&job_lock);
408    if (waiting > 0) {
409       pthread_cond_broadcast(&resource_wait);
410    }
411    V(mutex);
412 #endif
413 }
414
415 /*
416  * Get or create a Client record for this Job
417  */
418 int get_or_create_client_record(JCR *jcr)
419 {
420    CLIENT_DBR cr;
421
422    memset(&cr, 0, sizeof(cr));
423    bstrncpy(cr.Name, jcr->client->hdr.name, sizeof(cr.Name));
424    cr.AutoPrune = jcr->client->AutoPrune;
425    cr.FileRetention = jcr->client->FileRetention;
426    cr.JobRetention = jcr->client->JobRetention;
427    if (jcr->client_name) {
428       free_pool_memory(jcr->client_name);
429    }
430    jcr->client_name = get_memory(strlen(jcr->client->hdr.name) + 1);
431    strcpy(jcr->client_name, jcr->client->hdr.name);
432    if (!db_create_client_record(jcr, jcr->db, &cr)) {
433       Jmsg(jcr, M_FATAL, 0, _("Could not create Client record. ERR=%s\n"), 
434          db_strerror(jcr->db));
435       return 0;
436    }
437    jcr->jr.ClientId = cr.ClientId;
438    if (cr.Uname[0]) {
439       if (jcr->client_uname) {
440          free_pool_memory(jcr->client_uname);
441       }
442       jcr->client_uname = get_memory(strlen(cr.Uname) + 1);
443       strcpy(jcr->client_uname, cr.Uname);
444    }
445    Dmsg2(100, "Created Client %s record %d\n", jcr->client->hdr.name, 
446       jcr->jr.ClientId);
447    return 1;
448 }
449
450
451 /*
452  * Write status and such in DB
453  */
454 void update_job_end_record(JCR *jcr)
455 {
456    if (jcr->jr.EndTime == 0) {
457       jcr->jr.EndTime = time(NULL);
458    }
459    jcr->end_time = jcr->jr.EndTime;
460    jcr->jr.JobId = jcr->JobId;
461    jcr->jr.JobStatus = jcr->JobStatus;
462    jcr->jr.JobFiles = jcr->JobFiles;
463    jcr->jr.JobBytes = jcr->JobBytes;
464    jcr->jr.VolSessionId = jcr->VolSessionId;
465    jcr->jr.VolSessionTime = jcr->VolSessionTime;
466    if (!db_update_job_end_record(jcr, jcr->db, &jcr->jr)) {
467       Jmsg(jcr, M_WARNING, 0, _("Error updating job record. %s"), 
468          db_strerror(jcr->db));
469    }
470 }
471
472 /*
473  * Takes base_name and appends (unique) current
474  *   date and time to form unique job name.
475  *
476  *  Returns: unique job name in jcr->Job
477  *    date/time in jcr->start_time
478  */
479 void create_unique_job_name(JCR *jcr, char *base_name)
480 {
481    /* Job start mutex */
482    static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
483    static time_t last_start_time = 0;
484    time_t now;
485    struct tm tm;
486    char dt[MAX_TIME_LENGTH];
487    char name[MAX_NAME_LENGTH];
488    char *p;
489
490    /* Guarantee unique start time -- maximum one per second, and
491     * thus unique Job Name 
492     */
493    P(mutex);                          /* lock creation of jobs */
494    now = time(NULL);
495    while (now == last_start_time) {
496       bmicrosleep(0, 500000);
497       now = time(NULL);
498    }
499    last_start_time = now;
500    V(mutex);                          /* allow creation of jobs */
501    jcr->start_time = now;
502    /* Form Unique JobName */
503    localtime_r(&now, &tm);
504    /* Use only characters that are permitted in Windows filenames */
505    strftime(dt, sizeof(dt), "%Y-%m-%d_%H.%M.%S", &tm); 
506    bstrncpy(name, base_name, sizeof(name));
507    name[sizeof(name)-22] = 0;          /* truncate if too long */
508    sprintf(jcr->Job, "%s.%s", name, dt); /* add date & time */
509    /* Convert spaces into underscores */
510    for (p=jcr->Job; *p; p++) {
511       if (*p == ' ') {
512          *p = '_';
513       }
514    }
515 }
516
517 /*
518  * Free the Job Control Record if no one is still using it.
519  *  Called from main free_jcr() routine in src/lib/jcr.c so
520  *  that we can do our Director specific cleanup of the jcr.
521  */
522 void dird_free_jcr(JCR *jcr)
523 {
524    Dmsg0(200, "Start dird free_jcr\n");
525
526    if (jcr->file_bsock) {
527       Dmsg0(200, "Close File bsock\n");
528       bnet_close(jcr->file_bsock);
529    }
530    if (jcr->store_bsock) {
531       Dmsg0(200, "Close Store bsock\n");
532       bnet_close(jcr->store_bsock);
533    }
534    if (jcr->fname) {  
535       Dmsg0(200, "Free JCR fname\n");
536       free_pool_memory(jcr->fname);
537    }
538    if (jcr->stime) {
539       Dmsg0(200, "Free JCR stime\n");
540       free_pool_memory(jcr->stime);
541    }
542    if (jcr->db) {
543       Dmsg0(200, "Close DB\n");
544       db_close_database(jcr, jcr->db);
545    }
546    if (jcr->RestoreWhere) {
547       free(jcr->RestoreWhere);
548    }
549    if (jcr->RestoreBootstrap) {
550       free(jcr->RestoreBootstrap);
551    }
552    if (jcr->client_uname) {
553       free_pool_memory(jcr->client_uname);
554    }
555    Dmsg0(200, "End dird free_jcr\n");
556 }
557
558 /*
559  * Set some defaults in the JCR necessary to
560  * run. These items are pulled from the job
561  * definition as defaults, but can be overridden
562  * later either by the Run record in the Schedule resource,
563  * or by the Console program.
564  */
565 void set_jcr_defaults(JCR *jcr, JOB *job)
566 {
567    jcr->job = job;
568    jcr->JobType = job->JobType;
569    jcr->JobLevel = job->level;
570    jcr->store = job->storage;
571    jcr->client = job->client;
572    if (jcr->client_name) {
573       free_pool_memory(jcr->client_name);
574    }
575    jcr->client_name = get_memory(strlen(jcr->client->hdr.name) + 1);
576    strcpy(jcr->client_name, jcr->client->hdr.name);
577    jcr->pool = job->pool;
578    jcr->catalog = job->client->catalog;
579    jcr->fileset = job->fileset;
580    jcr->messages = job->messages; 
581    if (jcr->RestoreBootstrap) {
582       free(jcr->RestoreBootstrap);
583    }
584    /* This can be overridden by Console program */
585    if (job->RestoreBootstrap) {
586       jcr->RestoreBootstrap = bstrdup(job->RestoreBootstrap);
587    }
588    /* If no default level given, set one */
589    if (jcr->JobLevel == 0) {
590       switch (jcr->JobType) {
591       case JT_VERIFY:
592          jcr->JobLevel = L_VERIFY_CATALOG;
593          break;
594       case JT_BACKUP:
595          jcr->JobLevel = L_INCREMENTAL;
596          break;
597       case JT_RESTORE:
598       case JT_ADMIN:
599          jcr->JobLevel = L_FULL;
600          break;
601       default:
602          break;
603       }
604    }
605 }
606
607 /*
608  * Edit codes into Run command
609  *  %% = %
610  *  %c = Client's name
611  *  %d = Director's name
612  *  %i = JobId
613  *  %e = Job Exit
614  *  %j = Job
615  *  %l = Job Level
616  *  %n = Job name
617  *  %t = Job type
618  *
619  *  omsg = edited output message
620  *  imsg = input string containing edit codes (%x)
621  *
622  */
623 static char *edit_run_codes(JCR *jcr, char *omsg, char *imsg) 
624 {
625    char *p;
626    const char *str;
627    char add[20];
628
629    *omsg = 0;
630    Dmsg1(200, "edit_run_codes: %s\n", imsg);
631    for (p=imsg; *p; p++) {
632       if (*p == '%') {
633          switch (*++p) {
634          case '%':
635             str = "%";
636             break;
637          case 'c':
638             str = jcr->client_name;
639             if (!str) {
640                str = "";
641             }
642             break;
643          case 'd':
644             str = my_name;
645             break;
646          case 'e':
647             str = job_status_to_str(jcr->JobStatus);
648             break;
649          case 'i':
650             sprintf(add, "%d", jcr->JobId);
651             str = add;
652             break;
653          case 'j':                    /* Job */
654             str = jcr->Job;
655             break;
656          case 'l':
657             str = job_level_to_str(jcr->JobLevel);
658             break;
659          case 'n':
660             str = jcr->job->hdr.name;
661             break;
662          case 't':
663             str = job_type_to_str(jcr->JobType);
664             break;
665          default:
666             add[0] = '%';
667             add[1] = *p;
668             add[2] = 0;
669             str = add;
670             break;
671          }
672       } else {
673          add[0] = *p;
674          add[1] = 0;
675          str = add;
676       }
677       Dmsg1(200, "add_str %s\n", str);
678       pm_strcat(&omsg, (char *)str);
679       Dmsg1(200, "omsg=%s\n", omsg);
680    }
681    return omsg;
682 }