2 Bacula® - The Network Backup Solution
4 Copyright (C) 2003-2014 Free Software Foundation Europe e.V.
6 The main author of Bacula is Kern Sibbald, with contributions from many
7 others, a complete list can be found in the file AUTHORS.
9 You may use this file and others of this release according to the
10 license defined in the LICENSE file, which includes the Affero General
11 Public License, v3.0 ("AGPLv3") and some additional permissions and
12 terms pursuant to its AGPLv3 Section 7.
14 Bacula® is a registered trademark of Kern Sibbald.
17 * Bacula job queue routines.
19 * This code consists of three queues, the waiting_jobs
20 * queue, where jobs are initially queued, the ready_jobs
21 * queue, where jobs are placed when all the resources are
22 * allocated and they can immediately be run, and the
23 * running queue where jobs are placed when they are
26 * Kern Sibbald, July MMIII
29 * This code was adapted from the Bacula workq, which was
30 * adapted from "Programming with POSIX Threads", by
40 /* Forward referenced functions */
41 extern "C" void *jobq_server(void *arg);
42 extern "C" void *sched_wait(void *arg);
44 static int start_server(jobq_t *jq);
45 static bool acquire_resources(JCR *jcr);
46 static bool reschedule_job(JCR *jcr, jobq_t *jq, jobq_item_t *je);
47 static void dec_write_store(JCR *jcr);
50 * Initialize a job queue
52 * Returns: 0 on success
55 int jobq_init(jobq_t *jq, int threads, void *(*engine)(void *arg))
58 jobq_item_t *item = NULL;
60 if ((stat = pthread_attr_init(&jq->attr)) != 0) {
62 Jmsg1(NULL, M_ERROR, 0, _("pthread_attr_init: ERR=%s\n"), be.bstrerror(stat));
65 if ((stat = pthread_attr_setdetachstate(&jq->attr, PTHREAD_CREATE_DETACHED)) != 0) {
66 pthread_attr_destroy(&jq->attr);
69 if ((stat = pthread_mutex_init(&jq->mutex, NULL)) != 0) {
71 Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_init: ERR=%s\n"), be.bstrerror(stat));
72 pthread_attr_destroy(&jq->attr);
75 if ((stat = pthread_cond_init(&jq->work, NULL)) != 0) {
77 Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_init: ERR=%s\n"), be.bstrerror(stat));
78 pthread_mutex_destroy(&jq->mutex);
79 pthread_attr_destroy(&jq->attr);
83 jq->max_workers = threads; /* max threads to create */
84 jq->num_workers = 0; /* no threads yet */
85 jq->idle_workers = 0; /* no idle threads */
86 jq->engine = engine; /* routine to run */
87 jq->valid = JOBQ_VALID;
88 /* Initialize the job queues */
89 jq->waiting_jobs = New(dlist(item, &item->link));
90 jq->running_jobs = New(dlist(item, &item->link));
91 jq->ready_jobs = New(dlist(item, &item->link));
96 * Destroy the job queue
98 * Returns: 0 on success
101 int jobq_destroy(jobq_t *jq)
103 int stat, stat1, stat2;
105 if (jq->valid != JOBQ_VALID) {
109 jq->valid = 0; /* prevent any more operations */
112 * If any threads are active, wake them
114 if (jq->num_workers > 0) {
116 if (jq->idle_workers) {
117 if ((stat = pthread_cond_broadcast(&jq->work)) != 0) {
119 Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_broadcast: ERR=%s\n"), be.bstrerror(stat));
124 while (jq->num_workers > 0) {
125 if ((stat = pthread_cond_wait(&jq->work, &jq->mutex)) != 0) {
127 Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_wait: ERR=%s\n"), be.bstrerror(stat));
134 stat = pthread_mutex_destroy(&jq->mutex);
135 stat1 = pthread_cond_destroy(&jq->work);
136 stat2 = pthread_attr_destroy(&jq->attr);
137 delete jq->waiting_jobs;
138 delete jq->running_jobs;
139 delete jq->ready_jobs;
140 return (stat != 0 ? stat : (stat1 != 0 ? stat1 : stat2));
149 * Wait until schedule time arrives before starting. Normally
150 * this routine is only used for jobs started from the console
151 * for which the user explicitly specified a start time. Otherwise
152 * most jobs are put into the job queue only when their
153 * scheduled time arives.
156 void *sched_wait(void *arg)
158 JCR *jcr = ((wait_pkt *)arg)->jcr;
159 jobq_t *jq = ((wait_pkt *)arg)->jq;
161 set_jcr_in_tsd(INVALID_JCR);
162 Dmsg0(2300, "Enter sched_wait.\n");
164 time_t wtime = jcr->sched_time - time(NULL);
165 jcr->setJobStatus(JS_WaitStartTime);
166 /* Wait until scheduled time arrives */
168 Jmsg(jcr, M_INFO, 0, _("Job %s waiting %d seconds for scheduled start time.\n"),
171 /* Check every 30 seconds if canceled */
173 Dmsg3(2300, "Waiting on sched time, jobid=%d secs=%d use=%d\n",
174 jcr->JobId, wtime, jcr->use_count());
178 bmicrosleep(wtime, 0);
179 if (job_canceled(jcr)) {
182 wtime = jcr->sched_time - time(NULL);
184 Dmsg1(200, "resched use=%d\n", jcr->use_count());
186 free_jcr(jcr); /* we are done with jcr */
187 Dmsg0(2300, "Exit sched_wait\n");
192 * Add a job to the queue
193 * jq is a queue that was created with jobq_init
195 int jobq_add(jobq_t *jq, JCR *jcr)
198 jobq_item_t *item, *li;
199 bool inserted = false;
200 time_t wtime = jcr->sched_time - time(NULL);
204 if (!jcr->term_wait_inited) {
205 /* Initialize termination condition variable */
206 if ((stat = pthread_cond_init(&jcr->term_wait, NULL)) != 0) {
208 Jmsg1(jcr, M_FATAL, 0, _("Unable to init job cond variable: ERR=%s\n"), be.bstrerror(stat));
211 jcr->term_wait_inited = true;
214 Dmsg3(2300, "jobq_add jobid=%d jcr=0x%x use_count=%d\n", jcr->JobId, jcr, jcr->use_count());
215 if (jq->valid != JOBQ_VALID) {
216 Jmsg0(jcr, M_ERROR, 0, "Jobq_add queue not initialized.\n");
220 jcr->inc_use_count(); /* mark jcr in use by us */
221 Dmsg3(2300, "jobq_add jobid=%d jcr=0x%x use_count=%d\n", jcr->JobId, jcr, jcr->use_count());
222 if (!job_canceled(jcr) && wtime > 0) {
223 set_thread_concurrency(jq->max_workers + 2);
224 sched_pkt = (wait_pkt *)malloc(sizeof(wait_pkt));
225 sched_pkt->jcr = jcr;
227 stat = pthread_create(&id, &jq->attr, sched_wait, (void *)sched_pkt);
228 if (stat != 0) { /* thread not created */
230 Jmsg1(jcr, M_ERROR, 0, _("pthread_thread_create: ERR=%s\n"), be.bstrerror(stat));
237 if ((item = (jobq_item_t *)malloc(sizeof(jobq_item_t))) == NULL) {
238 free_jcr(jcr); /* release jcr */
243 /* While waiting in a queue this job is not attached to a thread */
244 set_jcr_in_tsd(INVALID_JCR);
245 if (job_canceled(jcr)) {
246 /* Add job to ready queue so that it is canceled quickly */
247 jq->ready_jobs->prepend(item);
248 Dmsg1(2300, "Prepended job=%d to ready queue\n", jcr->JobId);
250 /* Add this job to the wait queue in priority sorted order */
251 foreach_dlist(li, jq->waiting_jobs) {
252 Dmsg2(2300, "waiting item jobid=%d priority=%d\n",
253 li->jcr->JobId, li->jcr->JobPriority);
254 if (li->jcr->JobPriority > jcr->JobPriority) {
255 jq->waiting_jobs->insert_before(item, li);
256 Dmsg2(2300, "insert_before jobid=%d before waiting job=%d\n",
257 li->jcr->JobId, jcr->JobId);
262 /* If not jobs in wait queue, append it */
264 jq->waiting_jobs->append(item);
265 Dmsg1(2300, "Appended item jobid=%d to waiting queue\n", jcr->JobId);
269 /* Ensure that at least one server looks at the queue. */
270 stat = start_server(jq);
273 Dmsg0(2300, "Return jobq_add\n");
278 * Remove a job from the job queue. Used only by cancel_job().
279 * jq is a queue that was created with jobq_init
280 * work_item is an element of work
282 * Note, it is "removed" from the job queue.
283 * If you want to cancel it, you need to provide some external means
284 * of doing so (e.g. pthread_kill()).
286 int jobq_remove(jobq_t *jq, JCR *jcr)
292 Dmsg2(2300, "jobq_remove jobid=%d jcr=0x%x\n", jcr->JobId, jcr);
293 if (jq->valid != JOBQ_VALID) {
298 foreach_dlist(item, jq->waiting_jobs) {
299 if (jcr == item->jcr) {
306 Dmsg2(2300, "jobq_remove jobid=%d jcr=0x%x not in wait queue\n", jcr->JobId, jcr);
310 /* Move item to be the first on the list */
311 jq->waiting_jobs->remove(item);
312 jq->ready_jobs->prepend(item);
313 Dmsg2(2300, "jobq_remove jobid=%d jcr=0x%x moved to ready queue\n", jcr->JobId, jcr);
315 stat = start_server(jq);
318 Dmsg0(2300, "Return jobq_remove\n");
324 * Start the server thread if it isn't already running
326 static int start_server(jobq_t *jq)
332 * if any threads are idle, wake one.
333 * Actually we do a broadcast because on /lib/tls
334 * these signals seem to get lost from time to time.
336 if (jq->idle_workers > 0) {
337 Dmsg0(2300, "Signal worker to wake up\n");
338 if ((stat = pthread_cond_broadcast(&jq->work)) != 0) {
340 Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_signal: ERR=%s\n"), be.bstrerror(stat));
343 } else if (jq->num_workers < jq->max_workers) {
344 Dmsg0(2300, "Create worker thread\n");
345 /* No idle threads so create a new one */
346 set_thread_concurrency(jq->max_workers + 1);
348 if ((stat = pthread_create(&id, &jq->attr, jobq_server, (void *)jq)) != 0) {
351 Jmsg1(NULL, M_ERROR, 0, _("pthread_create: ERR=%s\n"), be.bstrerror(stat));
360 * This is the worker thread that serves the job queue.
361 * When all the resources are acquired for the job,
362 * it will call the user's engine.
365 void *jobq_server(void *arg)
367 struct timespec timeout;
368 jobq_t *jq = (jobq_t *)arg;
369 jobq_item_t *je; /* job entry in queue */
371 bool timedout = false;
374 set_jcr_in_tsd(INVALID_JCR);
375 Dmsg0(2300, "Start jobq_server\n");
382 Dmsg0(2300, "Top of for loop\n");
383 if (!work && !jq->quit) {
384 gettimeofday(&tv, &tz);
386 timeout.tv_sec = tv.tv_sec + 4;
390 * Wait 4 seconds, then if no more work, exit
392 Dmsg0(2300, "pthread_cond_timedwait()\n");
393 stat = pthread_cond_timedwait(&jq->work, &jq->mutex, &timeout);
394 if (stat == ETIMEDOUT) {
395 Dmsg0(2300, "timedwait timedout.\n");
398 } else if (stat != 0) {
399 /* This shouldn't happen */
400 Dmsg0(2300, "This shouldn't happen\n");
409 * If anything is in the ready queue, run it
411 Dmsg0(2300, "Checking ready queue.\n");
412 while (!jq->ready_jobs->empty() && !jq->quit) {
414 je = (jobq_item_t *)jq->ready_jobs->first();
416 jq->ready_jobs->remove(je);
417 if (!jq->ready_jobs->empty()) {
418 Dmsg0(2300, "ready queue not empty start server\n");
419 if (start_server(jq) != 0) {
425 jq->running_jobs->append(je);
427 /* Attach jcr to this thread while we run the job */
428 jcr->my_thread_id = pthread_self();
429 jcr->set_killable(true);
431 Dmsg1(2300, "Took jobid=%d from ready and appended to run\n", jcr->JobId);
433 /* Release job queue lock */
436 /* Call user's routine here */
437 Dmsg3(2300, "Calling user engine for jobid=%d use=%d stat=%c\n", jcr->JobId,
438 jcr->use_count(), jcr->JobStatus);
441 /* Job finished detach from thread */
442 remove_jcr_from_tsd(je->jcr);
443 je->jcr->set_killable(false);
445 /* Clear the threadid, probably not necessary */
446 memset(&jcr->my_thread_id, 0, sizeof(jcr->my_thread_id));
448 Dmsg2(2300, "Back from user engine jobid=%d use=%d.\n", jcr->JobId,
451 /* Reacquire job queue lock */
453 Dmsg0(200, "Done lock mutex after running job. Release locks.\n");
454 jq->running_jobs->remove(je);
456 * Release locks if acquired. Note, they will not have
457 * been acquired for jobs canceled before they were
458 * put into the ready queue.
460 if (jcr->acquired_resource_locks) {
462 dec_write_store(jcr);
464 jcr->client->NumConcurrentJobs--;
466 jcr->job->NumConcurrentJobs--;
467 jcr->acquired_resource_locks = false;
470 if (reschedule_job(jcr, jq, je)) {
471 continue; /* go look for more work */
474 /* Clean up and release old jcr */
475 Dmsg2(2300, "====== Termination job=%d use_cnt=%d\n", jcr->JobId, jcr->use_count());
476 jcr->SDJobStatus = 0;
477 V(jq->mutex); /* release internal lock */
479 free(je); /* release job entry */
480 P(jq->mutex); /* reacquire job queue lock */
483 * If any job in the wait queue can be run,
484 * move it to the ready queue
486 Dmsg0(2300, "Done check ready, now check wait queue.\n");
487 if (!jq->waiting_jobs->empty() && !jq->quit) {
489 bool running_allow_mix = false;
490 je = (jobq_item_t *)jq->waiting_jobs->first();
491 jobq_item_t *re = (jobq_item_t *)jq->running_jobs->first();
493 Priority = re->jcr->JobPriority;
494 Dmsg2(2300, "JobId %d is running. Look for pri=%d\n",
495 re->jcr->JobId, Priority);
496 running_allow_mix = true;
498 Dmsg2(2300, "JobId %d is also running with %s\n",
500 re->jcr->job->allow_mixed_priority ? "mix" : "no mix");
501 if (!re->jcr->job->allow_mixed_priority) {
502 running_allow_mix = false;
505 re = (jobq_item_t *)jq->running_jobs->next(re);
507 Dmsg1(2300, "The running job(s) %s mixing priorities.\n",
508 running_allow_mix ? "allow" : "don't allow");
510 Priority = je->jcr->JobPriority;
511 Dmsg1(2300, "No job running. Look for Job pri=%d\n", Priority);
514 * Walk down the list of waiting jobs and attempt
515 * to acquire the resources it needs.
518 /* je is current job item on the queue, jn is the next one */
520 jobq_item_t *jn = (jobq_item_t *)jq->waiting_jobs->next(je);
522 Dmsg4(2300, "Examining Job=%d JobPri=%d want Pri=%d (%s)\n",
523 jcr->JobId, jcr->JobPriority, Priority,
524 jcr->job->allow_mixed_priority ? "mix" : "no mix");
526 /* Take only jobs of correct Priority */
527 if (!(jcr->JobPriority == Priority
528 || (jcr->JobPriority < Priority &&
529 jcr->job->allow_mixed_priority && running_allow_mix))) {
530 jcr->setJobStatus(JS_WaitPriority);
534 if (!acquire_resources(jcr)) {
535 /* If resource conflict, job is canceled */
536 if (!job_canceled(jcr)) {
537 je = jn; /* point to next waiting job */
543 * Got all locks, now remove it from wait queue and append it
544 * to the ready queue. Note, we may also get here if the
545 * job was canceled. Once it is "run", it will quickly
548 jq->waiting_jobs->remove(je);
549 jq->ready_jobs->append(je);
550 Dmsg1(2300, "moved JobId=%d from wait to ready queue\n", je->jcr->JobId);
551 je = jn; /* Point to next waiting job */
556 Dmsg0(2300, "Done checking wait queue.\n");
558 * If no more ready work and we are asked to quit, then do it
560 if (jq->ready_jobs->empty() && jq->quit) {
562 if (jq->num_workers == 0) {
563 Dmsg0(2300, "Wake up destroy routine\n");
564 /* Wake up destroy routine if he is waiting */
565 pthread_cond_broadcast(&jq->work);
569 Dmsg0(2300, "Check for work request\n");
571 * If no more work requests, and we waited long enough, quit
573 Dmsg2(2300, "timedout=%d read empty=%d\n", timedout,
574 jq->ready_jobs->empty());
575 if (jq->ready_jobs->empty() && timedout) {
576 Dmsg0(2300, "break big loop\n");
581 work = !jq->ready_jobs->empty() || !jq->waiting_jobs->empty();
584 * If a job is waiting on a Resource, don't consume all
585 * the CPU time looping looking for work, and even more
586 * important, release the lock so that a job that has
587 * terminated can give us the resource.
590 bmicrosleep(2, 0); /* pause for 2 seconds */
592 /* Recompute work as something may have changed in last 2 secs */
593 work = !jq->ready_jobs->empty() || !jq->waiting_jobs->empty();
595 Dmsg1(2300, "Loop again. work=%d\n", work);
596 } /* end of big for loop */
598 Dmsg0(200, "unlock mutex\n");
600 Dmsg0(2300, "End jobq_server\n");
605 * Returns true if cleanup done and we should look for more work
607 static bool reschedule_job(JCR *jcr, jobq_t *jq, jobq_item_t *je)
609 bool resched = false;
611 * Reschedule the job if requested and possible
613 /* Basic condition is that more reschedule times remain */
614 if (jcr->job->RescheduleTimes == 0 ||
615 jcr->reschedule_count < jcr->job->RescheduleTimes) {
617 /* Check for failed jobs */
618 (jcr->job->RescheduleOnError &&
619 !jcr->is_JobStatus(JS_Terminated) &&
620 !jcr->is_JobStatus(JS_Canceled) &&
621 jcr->is_JobType(JT_BACKUP));
624 char dt[50], dt2[50];
627 * Reschedule this job by cleaning it up, but
628 * reuse the same JobId if possible.
630 time_t now = time(NULL);
631 jcr->reschedule_count++;
632 jcr->sched_time = now + jcr->job->RescheduleInterval;
633 bstrftime(dt, sizeof(dt), now);
634 bstrftime(dt2, sizeof(dt2), jcr->sched_time);
635 Dmsg4(2300, "Rescheduled Job %s to re-run in %d seconds.(now=%u,then=%u)\n", jcr->Job,
636 (int)jcr->job->RescheduleInterval, now, jcr->sched_time);
637 Jmsg(jcr, M_INFO, 0, _("Rescheduled Job %s at %s to re-run in %d seconds (%s).\n"),
638 jcr->Job, dt, (int)jcr->job->RescheduleInterval, dt2);
639 dird_free_jcr_pointers(jcr); /* partial cleanup old stuff */
641 jcr->setJobStatus(JS_WaitStartTime);
642 jcr->SDJobStatus = 0;
644 if (!allow_duplicate_job(jcr)) {
647 /* Only jobs with no output jobs can run on same JCR */
648 if (jcr->JobBytes == 0) {
649 Dmsg2(2300, "Requeue job=%d use=%d\n", jcr->JobId, jcr->use_count());
652 * Special test here since a Virtual Full gets marked
653 * as a Full, so we look at the resource record
655 if (jcr->wasVirtualFull) {
656 jcr->setJobLevel(L_VIRTUAL_FULL);
659 * When we are using the same jcr then make sure to reset
660 * RealEndTime back to zero.
662 jcr->jr.RealEndTime = 0;
663 jobq_add(jq, jcr); /* queue the job to run again */
665 free_jcr(jcr); /* release jcr */
666 free(je); /* free the job entry */
667 return true; /* we already cleaned up */
670 * Something was actually backed up, so we cannot reuse
671 * the old JobId or there will be database record
672 * conflicts. We now create a new job, copying the
673 * appropriate fields.
675 JCR *njcr = new_jcr(sizeof(JCR), dird_free_jcr);
676 set_jcr_defaults(njcr, jcr->job);
677 njcr->reschedule_count = jcr->reschedule_count;
678 njcr->sched_time = jcr->sched_time;
679 njcr->initial_sched_time = jcr->initial_sched_time;
681 * Special test here since a Virtual Full gets marked
682 * as a Full, so we look at the resource record
684 if (jcr->wasVirtualFull) {
685 njcr->setJobLevel(L_VIRTUAL_FULL);
687 njcr->setJobLevel(jcr->getJobLevel());
689 njcr->pool = jcr->pool;
690 njcr->run_pool_override = jcr->run_pool_override;
691 njcr->next_pool = jcr->next_pool;
692 njcr->run_next_pool_override = jcr->run_next_pool_override;
693 njcr->full_pool = jcr->full_pool;
694 njcr->run_full_pool_override = jcr->run_full_pool_override;
695 njcr->inc_pool = jcr->inc_pool;
696 njcr->run_inc_pool_override = jcr->run_inc_pool_override;
697 njcr->diff_pool = jcr->diff_pool;
698 njcr->JobStatus = -1;
699 njcr->setJobStatus(jcr->JobStatus);
701 copy_rstorage(njcr, jcr->rstorage, _("previous Job"));
706 copy_wstorage(njcr, jcr->wstorage, _("previous Job"));
710 njcr->messages = jcr->messages;
711 njcr->spool_data = jcr->spool_data;
712 njcr->write_part_after_job = jcr->write_part_after_job;
713 Dmsg0(2300, "Call to run new job\n");
715 run_job(njcr); /* This creates a "new" job */
716 free_jcr(njcr); /* release "new" jcr */
718 Dmsg0(2300, "Back from running new job.\n");
724 * See if we can acquire all the necessary resources for the job (JCR)
726 * Returns: true if successful
727 * false if resource failure
729 static bool acquire_resources(JCR *jcr)
731 bool skip_this_jcr = false;
733 jcr->acquired_resource_locks = false;
735 * Turning this code off is likely to cause some deadlocks,
736 * but we do not really have enough information here to
737 * know if this is really a deadlock (it may be a dual drive
738 * autochanger), and in principle, the SD reservation system
739 * should detect these deadlocks, so push the work off on it.
742 if (jcr->rstore && jcr->rstore == jcr->wstore) { /* possible deadlock */
743 Jmsg(jcr, M_FATAL, 0, _("Job canceled. Attempt to read and write same device.\n"
744 " Read storage \"%s\" (From %s) -- Write storage \"%s\" (From %s)\n"),
745 jcr->rstore->name(), jcr->rstore_source, jcr->wstore->name(), jcr->wstore_source);
746 jcr->setJobStatus(JS_Canceled);
751 Dmsg1(200, "Rstore=%s\n", jcr->rstore->name());
752 if (!inc_read_store(jcr)) {
753 Dmsg1(200, "Fail rncj=%d\n", jcr->rstore->NumConcurrentJobs);
754 jcr->setJobStatus(JS_WaitStoreRes);
760 Dmsg1(200, "Wstore=%s\n", jcr->wstore->name());
761 if (jcr->wstore->NumConcurrentJobs < jcr->wstore->MaxConcurrentJobs) {
762 jcr->wstore->NumConcurrentJobs++;
763 Dmsg1(200, "Inc wncj=%d\n", jcr->wstore->NumConcurrentJobs);
764 } else if (jcr->rstore) {
766 skip_this_jcr = true;
768 Dmsg1(200, "Fail wncj=%d\n", jcr->wstore->NumConcurrentJobs);
769 skip_this_jcr = true;
773 jcr->setJobStatus(JS_WaitStoreRes);
778 if (jcr->client->NumConcurrentJobs < jcr->client->MaxConcurrentJobs) {
779 jcr->client->NumConcurrentJobs++;
781 /* Back out previous locks */
782 dec_write_store(jcr);
784 jcr->setJobStatus(JS_WaitClientRes);
788 if (jcr->job->NumConcurrentJobs < jcr->job->MaxConcurrentJobs) {
789 jcr->job->NumConcurrentJobs++;
791 /* Back out previous locks */
792 dec_write_store(jcr);
795 jcr->client->NumConcurrentJobs--;
797 jcr->setJobStatus(JS_WaitJobRes);
801 jcr->acquired_resource_locks = true;
805 static pthread_mutex_t rstore_mutex = PTHREAD_MUTEX_INITIALIZER;
808 * Note: inc_read_store() and dec_read_store() are
809 * called from select_rstore() in src/dird/restore.c
811 bool inc_read_store(JCR *jcr)
814 if (jcr->rstore->NumConcurrentJobs < jcr->rstore->MaxConcurrentJobs &&
815 (jcr->getJobType() == JT_RESTORE ||
816 jcr->rstore->MaxConcurrentReadJobs == 0 ||
817 jcr->rstore->NumConcurrentReadJobs < jcr->rstore->MaxConcurrentReadJobs)) {
818 jcr->rstore->NumConcurrentReadJobs++;
819 jcr->rstore->NumConcurrentJobs++;
820 Dmsg1(200, "Inc rncj=%d\n", jcr->rstore->NumConcurrentJobs);
828 void dec_read_store(JCR *jcr)
832 jcr->rstore->NumConcurrentReadJobs--; /* back out rstore */
833 jcr->rstore->NumConcurrentJobs--; /* back out rstore */
834 Dmsg1(200, "Dec rncj=%d\n", jcr->rstore->NumConcurrentJobs);
836 ASSERT(jcr->rstore->NumConcurrentReadJobs >= 0);
837 ASSERT(jcr->rstore->NumConcurrentJobs >= 0);
841 static void dec_write_store(JCR *jcr)
844 jcr->wstore->NumConcurrentJobs--;
845 Dmsg1(200, "Dec wncj=%d\n", jcr->wstore->NumConcurrentJobs);
846 ASSERT(jcr->wstore->NumConcurrentJobs >= 0);