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