]> git.sur5r.net Git - bacula/bacula/blobdiff - bacula/src/dird/jobq.c
kes Update copyright date in program files, and for the most part
[bacula/bacula] / bacula / src / dird / jobq.c
index 4d79e77a3d2d67d271afda8394233aeb8009b7b7..7440fd74d3de5249323e1fbe293c82a50981d347 100755 (executable)
@@ -208,6 +208,16 @@ int jobq_add(jobq_t *jq, JCR *jcr)
    pthread_t id;
    wait_pkt *sched_pkt;
 
+   if (!jcr->term_wait_inited) { 
+      /* Initialize termination condition variable */
+      if ((stat = pthread_cond_init(&jcr->term_wait, NULL)) != 0) {
+         berrno be;
+         Jmsg1(jcr, M_FATAL, 0, _("Unable to init job cond variable: ERR=%s\n"), be.strerror(stat));
+         return stat;
+      }
+      jcr->term_wait_inited = true;
+   }                           
+                             
    Dmsg3(2300, "jobq_add jobid=%d jcr=0x%x use_count=%d\n", jcr->JobId, jcr, jcr->use_count());
    if (jq->valid != JOBQ_VALID) {
       Jmsg0(jcr, M_ERROR, 0, "Jobq_add queue not initialized.\n");
@@ -453,9 +463,17 @@ void *jobq_server(void *arg)
           *  put into the ready queue.
           */
          if (jcr->acquired_resource_locks) {
-            jcr->store->NumConcurrentJobs--;
+            if (jcr->rstore) {
+               jcr->rstore->NumConcurrentJobs = 0;
+               Dmsg1(200, "Dec rncj=%d\n", jcr->rstore->NumConcurrentJobs);
+            }
+            if (jcr->wstore) {
+               jcr->wstore->NumConcurrentJobs--;
+               Dmsg1(200, "Dec wncj=%d\n", jcr->wstore->NumConcurrentJobs);
+            }
             jcr->client->NumConcurrentJobs--;
             jcr->job->NumConcurrentJobs--;
+            jcr->acquired_resource_locks = false;
          }
 
          /*
@@ -466,26 +484,29 @@ void *jobq_server(void *arg)
              jcr->JobStatus != JS_Canceled &&
              jcr->job->RescheduleTimes > 0 &&
              jcr->JobType == JT_BACKUP &&
-             jcr->reschedule_count < jcr->job->RescheduleTimes) {
-             char dt[50];
+             (jcr->job->RescheduleTimes == 0 ||
+              jcr->reschedule_count < jcr->job->RescheduleTimes)) {
+             char dt[50], dt2[50];
 
              /*
               * Reschedule this job by cleaning it up, but
               *  reuse the same JobId if possible.
               */
+            time_t now = time(NULL);
             jcr->reschedule_count++;
-            jcr->sched_time = time(NULL) + jcr->job->RescheduleInterval;
-            Dmsg2(2300, "Rescheduled Job %s to re-run in %d seconds.\n", jcr->Job,
-               (int)jcr->job->RescheduleInterval);
-            bstrftime(dt, sizeof(dt), time(NULL));
-            Jmsg(jcr, M_INFO, 0, _("Rescheduled Job %s at %s to re-run in %d seconds.\n"),
-               jcr->Job, dt, (int)jcr->job->RescheduleInterval);
+            jcr->sched_time = now + jcr->job->RescheduleInterval;
+            bstrftime(dt, sizeof(dt), now);
+            bstrftime(dt2, sizeof(dt2), jcr->sched_time);
+            Dmsg4(2300, "Rescheduled Job %s to re-run in %d seconds.(now=%u,then=%u)\n", jcr->Job,
+                  (int)jcr->job->RescheduleInterval, now, jcr->sched_time);
+            Jmsg(jcr, M_INFO, 0, _("Rescheduled Job %s at %s to re-run in %d seconds (%s).\n"),
+                 jcr->Job, dt, (int)jcr->job->RescheduleInterval, dt2);
             dird_free_jcr_pointers(jcr);     /* partial cleanup old stuff */
-            jcr->JobStatus = JS_WaitStartTime;
+            jcr->JobStatus = -1;
+            set_jcr_job_status(jcr, JS_WaitStartTime);
             jcr->SDJobStatus = 0;
             if (jcr->JobBytes == 0) {
                Dmsg2(2300, "Requeue job=%d use=%d\n", jcr->JobId, jcr->use_count());
-               jcr->JobStatus = JS_WaitStartTime;
                V(jq->mutex);
                jobq_add(jq, jcr);     /* queue the job to run again */
                P(jq->mutex);
@@ -498,13 +519,24 @@ void *jobq_server(void *arg)
              *   the old JobId or there will be database record
              *   conflicts.  We now create a new job, copying the
              *   appropriate fields.
-             */
+             */           
             JCR *njcr = new_jcr(sizeof(JCR), dird_free_jcr);
             set_jcr_defaults(njcr, jcr->job);
             njcr->reschedule_count = jcr->reschedule_count;
+            njcr->sched_time = jcr->sched_time;
             njcr->JobLevel = jcr->JobLevel;
-            njcr->JobStatus = jcr->JobStatus;
-            copy_storage(njcr, jcr->storage);
+            njcr->JobStatus = -1;
+            set_jcr_job_status(njcr, jcr->JobStatus);
+            if (jcr->rstore) {
+               copy_rstorage(njcr, jcr->rstorage, _("previous Job"));
+            } else {
+               free_rstorage(njcr);
+            }
+            if (jcr->wstore) {
+               copy_wstorage(njcr, jcr->wstorage, _("previous Job"));
+            } else {
+               free_wstorage(njcr);
+            }
             njcr->messages = jcr->messages;
             Dmsg0(2300, "Call to run new job\n");
             V(jq->mutex);
@@ -567,7 +599,6 @@ void *jobq_server(void *arg)
             /* Got all locks, now remove it from wait queue and append it
              *   to the ready queue
              */
-            jcr->acquired_resource_locks = true;
             jq->waiting_jobs->remove(je);
             jq->ready_jobs->append(je);
             Dmsg1(2300, "moved JobId=%d from wait to ready queue\n", je->jcr->JobId);
@@ -634,26 +665,40 @@ static bool acquire_resources(JCR *jcr)
 {
    bool skip_this_jcr = false;
 
-   if (jcr->JobType == JT_RESTORE || jcr->JobType == JT_VERIFY) {
+   jcr->acquired_resource_locks = false;
+   if (jcr->rstore) {
+      Dmsg1(200, "Rstore=%s\n", jcr->rstore->name());
       /* 
        * Let only one Restore/verify job run at a time regardless
        *   of MaxConcurrentJobs.
        */ 
-      if (jcr->store->NumConcurrentJobs == 0) {
-         jcr->store->NumConcurrentJobs = 1;
+      if (jcr->rstore->NumConcurrentJobs == 0) {
+         jcr->rstore->NumConcurrentJobs = 1;
+         Dmsg0(200, "Set rncj=1\n");
       } else {
+         Dmsg1(200, "Fail rncj=%d\n", jcr->rstore->NumConcurrentJobs);
          set_jcr_job_status(jcr, JS_WaitStoreRes);
          return false;
       }
-   /* We are not doing a Restore or Verify */
-   } else if (jcr->store->NumConcurrentJobs == 0 &&
-              jcr->store->NumConcurrentJobs < jcr->store->MaxConcurrentJobs) {
-       /* Simple case, first job */
-       jcr->store->NumConcurrentJobs = 1;
-   } else if (jcr->store->NumConcurrentJobs < jcr->store->MaxConcurrentJobs) {
-       jcr->store->NumConcurrentJobs++;
-   } else {
-      skip_this_jcr = true;
+   }
+   
+   if (jcr->wstore) {
+      if (jcr->wstore->NumConcurrentJobs == 0 &&
+          jcr->wstore->NumConcurrentJobs < jcr->wstore->MaxConcurrentJobs) {
+         /* Simple case, first job */
+         jcr->wstore->NumConcurrentJobs = 1;
+         Dmsg0(200, "Set wncj=1\n");
+      } else if (jcr->wstore->NumConcurrentJobs < jcr->wstore->MaxConcurrentJobs) {
+         jcr->wstore->NumConcurrentJobs++;
+         Dmsg1(200, "Inc wncj=%d\n", jcr->wstore->NumConcurrentJobs);
+      } else if (jcr->rstore) {
+         jcr->rstore->NumConcurrentJobs = 0;      /* back out rstore */
+         Dmsg1(200, "Fail wncj=%d\n", jcr->wstore->NumConcurrentJobs);
+         skip_this_jcr = true;
+      } else {
+         Dmsg1(200, "Fail wncj=%d\n", jcr->wstore->NumConcurrentJobs);
+         skip_this_jcr = true;
+      }
    }
    if (skip_this_jcr) {
       set_jcr_job_status(jcr, JS_WaitStoreRes);
@@ -664,7 +709,14 @@ static bool acquire_resources(JCR *jcr)
       jcr->client->NumConcurrentJobs++;
    } else {
       /* Back out previous locks */
-      jcr->store->NumConcurrentJobs--;
+      if (jcr->wstore) {
+         jcr->wstore->NumConcurrentJobs--;
+         Dmsg1(200, "Dec wncj=%d\n", jcr->wstore->NumConcurrentJobs);
+      }
+      if (jcr->rstore) {
+         jcr->rstore->NumConcurrentJobs = 0;
+         Dmsg1(200, "Dec rncj=%d\n", jcr->rstore->NumConcurrentJobs);
+      }
       set_jcr_job_status(jcr, JS_WaitClientRes);
       return false;
    }
@@ -672,12 +724,22 @@ static bool acquire_resources(JCR *jcr)
       jcr->job->NumConcurrentJobs++;
    } else {
       /* Back out previous locks */
-      jcr->store->NumConcurrentJobs--;
+      if (jcr->wstore) {
+         jcr->wstore->NumConcurrentJobs--;
+         Dmsg1(200, "Dec wncj=%d\n", jcr->wstore->NumConcurrentJobs);
+      }
+      if (jcr->rstore) {
+         jcr->rstore->NumConcurrentJobs = 0;
+         Dmsg1(200, "Dec rncj=%d\n", jcr->rstore->NumConcurrentJobs);
+      }
       jcr->client->NumConcurrentJobs--;
       set_jcr_job_status(jcr, JS_WaitJobRes);
       return false;
    }
    /* Check actual device availability */
    /* ***FIXME****/
+
+
+   jcr->acquired_resource_locks = true;
    return true;
 }