if (jq->valid != JOBQ_VALID) {
return EINVAL;
}
- if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
- berrno be;
- Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_lock: ERR=%s\n"), be.bstrerror(stat));
- return stat;
- }
+ P(jq->mutex);
jq->valid = 0; /* prevent any more operations */
/*
if ((stat = pthread_cond_broadcast(&jq->work)) != 0) {
berrno be;
Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_broadcast: ERR=%s\n"), be.bstrerror(stat));
- pthread_mutex_unlock(&jq->mutex);
+ V(jq->mutex);
return stat;
}
}
if ((stat = pthread_cond_wait(&jq->work, &jq->mutex)) != 0) {
berrno be;
Jmsg1(NULL, M_ERROR, 0, _("pthread_cond_wait: ERR=%s\n"), be.bstrerror(stat));
- pthread_mutex_unlock(&jq->mutex);
+ V(jq->mutex);
return stat;
}
}
}
- if ((stat = pthread_mutex_unlock(&jq->mutex)) != 0) {
- berrno be;
- Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_unlock: ERR=%s\n"), be.bstrerror(stat));
- return stat;
- }
+ V(jq->mutex);
stat = pthread_mutex_destroy(&jq->mutex);
stat1 = pthread_cond_destroy(&jq->work);
stat2 = pthread_attr_destroy(&jq->attr);
JCR *jcr = ((wait_pkt *)arg)->jcr;
jobq_t *jq = ((wait_pkt *)arg)->jq;
+ set_jcr_in_tsd(jcr);
Dmsg0(2300, "Enter sched_wait.\n");
free(arg);
time_t wtime = jcr->sched_time - time(NULL);
return stat;
}
- if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
- berrno be;
- Jmsg1(jcr, M_ERROR, 0, _("pthread_mutex_lock: ERR=%s\n"), be.bstrerror(stat));
- free_jcr(jcr); /* release jcr */
- return stat;
- }
+ P(jq->mutex);
if ((item = (jobq_item_t *)malloc(sizeof(jobq_item_t))) == NULL) {
free_jcr(jcr); /* release jcr */
}
item->jcr = jcr;
+ /* While waiting in a queue this job is not attached to a thread */
+ set_jcr_in_tsd(INVALID_JCR);
if (job_canceled(jcr)) {
/* Add job to ready queue so that it is canceled quickly */
jq->ready_jobs->prepend(item);
/* Ensure that at least one server looks at the queue. */
stat = start_server(jq);
- pthread_mutex_unlock(&jq->mutex);
+ V(jq->mutex);
Dmsg0(2300, "Return jobq_add\n");
return stat;
}
return EINVAL;
}
- if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
- berrno be;
- Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_lock: ERR=%s\n"), be.bstrerror(stat));
- return stat;
- }
-
+ P(jq->mutex);
foreach_dlist(item, jq->waiting_jobs) {
if (jcr == item->jcr) {
found = true;
}
}
if (!found) {
- pthread_mutex_unlock(&jq->mutex);
+ V(jq->mutex);
Dmsg2(2300, "jobq_remove jobid=%d jcr=0x%x not in wait queue\n", jcr->JobId, jcr);
return EINVAL;
}
stat = start_server(jq);
- pthread_mutex_unlock(&jq->mutex);
+ V(jq->mutex);
Dmsg0(2300, "Return jobq_remove\n");
return stat;
}
Dmsg0(2300, "Create worker thread\n");
/* No idle threads so create a new one */
set_thread_concurrency(jq->max_workers + 1);
+ jq->num_workers++;
if ((stat = pthread_create(&id, &jq->attr, jobq_server, (void *)jq)) != 0) {
berrno be;
+ jq->num_workers--;
Jmsg1(NULL, M_ERROR, 0, _("pthread_create: ERR=%s\n"), be.bstrerror(stat));
return stat;
}
bool timedout = false;
bool work = true;
+ set_jcr_in_tsd(INVALID_JCR);
Dmsg0(2300, "Start jobq_server\n");
- if ((stat = pthread_mutex_lock(&jq->mutex)) != 0) {
- berrno be;
- Jmsg1(NULL, M_ERROR, 0, _("pthread_mutex_lock: ERR=%s\n"), be.bstrerror(stat));
- return NULL;
- }
- jq->num_workers++;
+ P(jq->mutex);
for (;;) {
struct timeval tv;
/* This shouldn't happen */
Dmsg0(2300, "This shouldn't happen\n");
jq->num_workers--;
- pthread_mutex_unlock(&jq->mutex);
+ V(jq->mutex);
return NULL;
}
break;
Dmsg0(2300, "ready queue not empty start server\n");
if (start_server(jq) != 0) {
jq->num_workers--;
- pthread_mutex_unlock(&jq->mutex);
+ V(jq->mutex);
return NULL;
}
}
jq->running_jobs->append(je);
+
+ /* Attach jcr to this thread while we run the job */
set_jcr_in_tsd(jcr);
Dmsg1(2300, "Took jobid=%d from ready and appended to run\n", jcr->JobId);
jcr->use_count());
jq->engine(je->jcr);
+ /* Job finished detach from thread */
+ set_jcr_in_tsd(INVALID_JCR);
+
Dmsg2(2300, "Back from user engine jobid=%d use=%d.\n", jcr->JobId,
jcr->use_count());
if (re) {
Priority = re->jcr->JobPriority;
Dmsg2(2300, "JobId %d is running. Look for pri=%d\n",
- re->jcr->JobId, Priority);
- running_allow_mix = true;
- for ( ; re; ) {
- Dmsg2(2300, "JobId %d is also running with %s\n",
- re->jcr->JobId,
- re->jcr->job->allow_mixed_priority ? "mix" : "no mix");
- if (!re->jcr->job->allow_mixed_priority) {
- running_allow_mix = false;
- break;
- }
- re = (jobq_item_t *)jq->running_jobs->next(re);
- }
+ re->jcr->JobId, Priority);
+ running_allow_mix = true;
+ for ( ; re; ) {
+ Dmsg2(2300, "JobId %d is also running with %s\n",
+ re->jcr->JobId,
+ re->jcr->job->allow_mixed_priority ? "mix" : "no mix");
+ if (!re->jcr->job->allow_mixed_priority) {
+ running_allow_mix = false;
+ break;
+ }
+ re = (jobq_item_t *)jq->running_jobs->next(re);
+ }
Dmsg1(2300, "The running job(s) %s mixing priorities.\n",
- running_allow_mix ? "allow" : "don't allow");
+ running_allow_mix ? "allow" : "don't allow");
} else {
Priority = je->jcr->JobPriority;
Dmsg1(2300, "No job running. Look for Job pri=%d\n", Priority);
jobq_item_t *jn = (jobq_item_t *)jq->waiting_jobs->next(je);
Dmsg4(2300, "Examining Job=%d JobPri=%d want Pri=%d (%s)\n",
- jcr->JobId, jcr->JobPriority, Priority,
- jcr->job->allow_mixed_priority ? "mix" : "no mix");
+ jcr->JobId, jcr->JobPriority, Priority,
+ jcr->job->allow_mixed_priority ? "mix" : "no mix");
/* Take only jobs of correct Priority */
if (!(jcr->JobPriority == Priority
- || (jcr->JobPriority < Priority &&
- jcr->job->allow_mixed_priority && running_allow_mix))) {
+ || (jcr->JobPriority < Priority &&
+ jcr->job->allow_mixed_priority && running_allow_mix))) {
set_jcr_job_status(jcr, JS_WaitPriority);
break;
}
bool skip_this_jcr = false;
jcr->acquired_resource_locks = false;
+/*
+ * Turning this code off is likely to cause some deadlocks,
+ * but we do not really have enough information here to
+ * know if this is really a deadlock (it may be a dual drive
+ * autochanger), and in principle, the SD reservation system
+ * should detect these deadlocks, so push the work off on is.
+ */
+#ifdef xxx
if (jcr->rstore && jcr->rstore == jcr->wstore) { /* possible deadlock */
Jmsg(jcr, M_FATAL, 0, _("Job canceled. Attempt to read and write same device.\n"
" Read storage \"%s\" (From %s) -- Write storage \"%s\" (From %s)\n"),
set_jcr_job_status(jcr, JS_Canceled);
return false;
}
+#endif
if (jcr->rstore) {
Dmsg1(200, "Rstore=%s\n", jcr->rstore->name());
if (jcr->rstore->NumConcurrentJobs < jcr->rstore->MaxConcurrentJobs) {
projects / bacula / bacula / blobdiff