src/dird/jobq.c
int jobq_add(jobq_t *jq, JCR *jcr)
{
   int stat;
   jobq_item_t *item, *li;
   bool inserted = false;
   time_t wtime = jcr->sched_time - time(NULL);
   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.bstrerror(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");
      return EINVAL;
   }

   jcr->inc_use_count();                 /* mark jcr in use by us */
   Dmsg3(2300, "jobq_add jobid=%d jcr=0x%x use_count=%d\n", jcr->JobId, jcr, jcr->use_count());
   if (!job_canceled(jcr) && wtime > 0) {
      set_thread_concurrency(jq->max_workers + 2);
      sched_pkt = (wait_pkt *)malloc(sizeof(wait_pkt));
      sched_pkt->jcr = jcr;
      sched_pkt->jq = jq;
      stat = pthread_create(&id, &jq->attr, sched_wait, (void *)sched_pkt);
      if (stat != 0) {                /* thread not created */
         berrno be;
         Jmsg1(jcr, M_ERROR, 0, _("pthread_thread_create: ERR=%s\n"), be.bstrerror(stat));
      }
      return stat;
   }

   P(jq->mutex);

   if ((item = (jobq_item_t *)malloc(sizeof(jobq_item_t))) == NULL) {
      free_jcr(jcr);                    /* release jcr */
      return ENOMEM;
   }
   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);
      Dmsg1(2300, "Prepended job=%d to ready queue\n", jcr->JobId);
   } else {
      /* Add this job to the wait queue in priority sorted order */
      foreach_dlist(li, jq->waiting_jobs) {
         Dmsg2(2300, "waiting item jobid=%d priority=%d\n",
            li->jcr->JobId, li->jcr->JobPriority);
         if (li->jcr->JobPriority > jcr->JobPriority) {
            jq->waiting_jobs->insert_before(item, li);
            Dmsg2(2300, "insert_before jobid=%d before waiting job=%d\n",
               li->jcr->JobId, jcr->JobId);
            inserted = true;
            break;
         }
      }
      /* If not jobs in wait queue, append it */
      if (!inserted) {
         jq->waiting_jobs->append(item);
         Dmsg1(2300, "Appended item jobid=%d to waiting queue\n", jcr->JobId);
      }
   }

   /* Ensure that at least one server looks at the queue. */
   stat = start_server(jq);

   V(jq->mutex);
   Dmsg0(2300, "Return jobq_add\n");
   return stat;
}
/*
 *  Add a job to the queue
 *    jq is a queue that was created with jobq_init
 */
Add a job to the queue
jq is a queue that was created with jobq_init
Add a job to the queue jq is a queue that was created with jobq_init
int jobq_add(jobq_t *jq, JCR *jcr)
{
   int stat;
   jobq_item_t *item, *li;
   bool inserted = false;
   time_t wtime = jcr->sched_time - time(NULL);
   pthread_t id;
   wait_pkt *sched_pkt;
   if (!jcr->term_wait_inited) {
      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.bstrerror(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");
      return EINVAL;
   }
   jcr->inc_use_count();                 
   Dmsg3(2300, "jobq_add jobid=%d jcr=0x%x use_count=%d\n", jcr->JobId, jcr, jcr->use_count());
   if (!job_canceled(jcr) && wtime > 0) {
      set_thread_concurrency(jq->max_workers + 2);
      sched_pkt = (wait_pkt *)malloc(sizeof(wait_pkt));
      sched_pkt->jcr = jcr;
      sched_pkt->jq = jq;
      stat = pthread_create(&id, &jq->attr, sched_wait, (void *)sched_pkt);
      if (stat != 0) {                
         berrno be;
         Jmsg1(jcr, M_ERROR, 0, _("pthread_thread_create: ERR=%s\n"), be.bstrerror(stat));
      }
      return stat;
   }
   P(jq->mutex);
   if ((item = (jobq_item_t *)malloc(sizeof(jobq_item_t))) == NULL) {
      free_jcr(jcr);                    
      return ENOMEM;
   }
   item->jcr = jcr;
   set_jcr_in_tsd(INVALID_JCR);
   if (job_canceled(jcr)) {
      jq->ready_jobs->prepend(item);
      Dmsg1(2300, "Prepended job=%d to ready queue\n", jcr->JobId);
   } else {
      foreach_dlist(li, jq->waiting_jobs) {
         Dmsg2(2300, "waiting item jobid=%d priority=%d\n",
            li->jcr->JobId, li->jcr->JobPriority);
         if (li->jcr->JobPriority > jcr->JobPriority) {
            jq->waiting_jobs->insert_before(item, li);
            Dmsg2(2300, "insert_before jobid=%d before waiting job=%d\n",
               li->jcr->JobId, jcr->JobId);
            inserted = true;
            break;
         }
      }
      if (!inserted) {
         jq->waiting_jobs->append(item);
         Dmsg1(2300, "Appended item jobid=%d to waiting queue\n", jcr->JobId);
      }
   }
   stat = start_server(jq);
   V(jq->mutex);
   Dmsg0(2300, "Return jobq_add\n");
   return stat;
}
int jobq_add ( jobq_t * jq , JCR * jcr ) { int stat ; jobq_item_t * item , * li ; bool inserted = false ; time_t wtime = jcr -> sched_time - time ( NULL ) ; pthread_t id ; wait_pkt * sched_pkt ; if ( ! jcr -> term_wait_inited ) { if ( ( stat = pthread_cond_init ( & jcr -> term_wait , NULL ) ) != 0 ) { berrno be ; Jmsg1 ( jcr , M_FATAL , 0 , _ ( " \n " ) , be . bstrerror ( stat ) ) ; return stat ; } jcr -> term_wait_inited = true ; } Dmsg3 ( 2300 , " \n " , jcr -> JobId , jcr , jcr -> use_count ( ) ) ; if ( jq -> valid != JOBQ_VALID ) { Jmsg0 ( jcr , M_ERROR , 0 , " \n " ) ; return EINVAL ; } jcr -> inc_use_count ( ) ; Dmsg3 ( 2300 , " \n " , jcr -> JobId , jcr , jcr -> use_count ( ) ) ; if ( ! job_canceled ( jcr ) && wtime > 0 ) { set_thread_concurrency ( jq -> max_workers + 2 ) ; sched_pkt = ( wait_pkt * ) malloc ( sizeof ( wait_pkt ) ) ; sched_pkt -> jcr = jcr ; sched_pkt -> jq = jq ; stat = pthread_create ( & id , & jq -> attr , sched_wait , ( void * ) sched_pkt ) ; if ( stat != 0 ) { berrno be ; Jmsg1 ( jcr , M_ERROR , 0 , _ ( " \n " ) , be . bstrerror ( stat ) ) ; } return stat ; } P ( jq -> mutex ) ; if ( ( item = ( jobq_item_t * ) malloc ( sizeof ( jobq_item_t ) ) ) == NULL ) { free_jcr ( jcr ) ; return ENOMEM ; } item -> jcr = jcr ; set_jcr_in_tsd ( INVALID_JCR ) ; if ( job_canceled ( jcr ) ) { jq -> ready_jobs -> prepend ( item ) ; Dmsg1 ( 2300 , " \n " , jcr -> JobId ) ; } else { foreach_dlist ( li , jq -> waiting_jobs ) { Dmsg2 ( 2300 , " \n " , li -> jcr -> JobId , li -> jcr -> JobPriority ) ; if ( li -> jcr -> JobPriority > jcr -> JobPriority ) { jq -> waiting_jobs -> insert_before ( item , li ) ; Dmsg2 ( 2300 , " \n " , li -> jcr -> JobId , jcr -> JobId ) ; inserted = true ; break ; } } if ( ! inserted ) { jq -> waiting_jobs -> append ( item ) ; Dmsg1 ( 2300 , " \n " , jcr -> JobId ) ; } } stat = start_server ( jq ) ; V ( jq -> mutex ) ; Dmsg0 ( 2300 , " \n " ) ; return stat ; }
Add a job to the queue
    jq is a queue that was created with jobq_init
Add a job to the queue jq is a queue that was created with jobq_init
/* Initialize termination condition variable */ /* mark jcr in use by us */ /* thread not created */ /* release jcr */ /* While waiting in a queue this job is not attached to a thread */ /* Add job to ready queue so that it is canceled quickly */ /* Add this job to the wait queue in priority sorted order */ /* If not jobs in wait queue, append it */ /* Ensure that at least one server looks at the queue. */