自学教程：C++ svr_setjobstate函数代码示例

END_TESTSTART_TEST(svr_setjobstate_test)  {  struct job test_job;  int result = PBSE_NONE;  memset(&test_job, 0, sizeof(test_job));  /*initialize_globals*/  server.sv_qs_mutex = (pthread_mutex_t *)calloc(1, sizeof(pthread_mutex_t));  server.sv_attr_mutex = (pthread_mutex_t *)calloc(1, sizeof(pthread_mutex_t));  server.sv_jobstates_mutex = (pthread_mutex_t *)calloc(1, sizeof(pthread_mutex_t));  pthread_mutex_init(server.sv_qs_mutex,NULL);  pthread_mutex_init(server.sv_attr_mutex,NULL);  pthread_mutex_init(server.sv_jobstates_mutex,NULL);  result = svr_setjobstate(NULL, 0, 0, 0);  fail_unless(result == PBSE_BAD_PARAMETER, "NULL input pointer fail");  result = svr_setjobstate(&test_job, 0, 0, 0);  fail_unless(result == PBSE_NONE, "svr_setjobstate fail");  result = svr_setjobstate(&test_job, 1, 2, 3);  fail_unless(result == PBSE_NONE, "svr_setjobstate fail");  }

void update_array_statuses()  {  job_array      *pa;  job            *pjob;  int             iter = -1;  unsigned int    running;  unsigned int    queued;  unsigned int    complete;  char            log_buf[LOCAL_LOG_BUF_SIZE];  char            jobid[PBS_MAXSVRJOBID+1];  while ((pa = next_array(&iter)) != NULL)    {    running  = pa->ai_qs.jobs_running;    complete = pa->ai_qs.num_failed + pa->ai_qs.num_successful;    queued   = pa->ai_qs.num_jobs - running - complete;        if (LOGLEVEL >= 7)      {      sprintf(log_buf, "%s: unlocking ai_mutex", __func__);      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buf);      }    strcpy(jobid, pa->ai_qs.parent_id);    unlock_ai_mutex(pa, __func__, "1", LOGLEVEL);        if ((pjob = svr_find_job(jobid, TRUE)) != NULL)      {      if (running > 0)        {        svr_setjobstate(pjob, JOB_STATE_RUNNING, pjob->ji_qs.ji_substate, FALSE);        }      else if ((complete > 0) &&                (queued == 0))        {        svr_setjobstate(pjob, JOB_STATE_COMPLETE, pjob->ji_qs.ji_substate, FALSE);        }      else         {        /* default to just calling the array queued */        svr_setjobstate(pjob, JOB_STATE_QUEUED, pjob->ji_qs.ji_substate, FALSE);        }      unlock_ji_mutex(pjob, __func__, "2", LOGLEVEL);      }    } /* END for each array */  } /* END update_array_statuses() */

void post_rerun(  batch_request *preq)  {  int   newstate;  int   newsub;  job  *pjob;  char  log_buf[LOCAL_LOG_BUF_SIZE];  if (preq == NULL)    return;  if (preq->rq_reply.brp_code != 0)    {    sprintf(log_buf, "rerun signal reject by mom: %s - %d", preq->rq_ind.rq_signal.rq_jid, preq->rq_reply.brp_code);    log_event(PBSEVENT_JOB,PBS_EVENTCLASS_JOB,__func__,log_buf);    if ((pjob = svr_find_job(preq->rq_ind.rq_signal.rq_jid, FALSE)))      {      mutex_mgr job_mutex(pjob->ji_mutex, true);            svr_evaljobstate(*pjob, newstate, newsub, 1);      svr_setjobstate(pjob, newstate, newsub, FALSE);      }    }  return;  }  /* END post_rerun() */

/* * release_job - releases the hold on job j * @param j - the job to modify * @return 0 if successful, a PBS error on failure */int release_job(  struct batch_request *preq, /* I */  void                 *j)    /* I/O */  {  long           old_hold;  int            rc = 0;  int            newstate;  int            newsub;  char          *pset;  job           *pjob = (job *)j;  char           log_buf[LOCAL_LOG_BUF_SIZE];  pbs_attribute  temphold;  /* cannot do anything until we decode the holds to be set */  if ((rc = get_hold(&preq->rq_ind.rq_hold.rq_orig.rq_attr, &pset, &temphold)) != 0)    {    return(rc);    }  /* if other than HOLD_u is being released, must have privil */  if ((rc = chk_hold_priv(temphold.at_val.at_long, preq->rq_perm)) != 0)    {    return(rc);    }  /* unset the hold */  old_hold = pjob->ji_wattr[JOB_ATR_hold].at_val.at_long;  if ((rc = job_attr_def[JOB_ATR_hold].at_set(&pjob->ji_wattr[JOB_ATR_hold], &temphold, DECR)))    {    return(rc);    }  /* everything went well, if holds changed, update the job state */  if (old_hold != pjob->ji_wattr[JOB_ATR_hold].at_val.at_long)    {    pjob->ji_modified = 1; /* indicates attributes changed */    svr_evaljobstate(pjob, &newstate, &newsub, 0);    svr_setjobstate(pjob, newstate, newsub, FALSE); /* saves job */    }  sprintf(log_buf, msg_jobholdrel,    pset,    preq->rq_user,    preq->rq_host);  log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buf);  return(rc);  } /* END release_job() */

static int svr_stagein(  job                  *pjob,     /* I */  struct batch_request *preq,     /* I */  int                   state,    /* I */  int                   substate) /* I */  {  struct batch_request *momreq = 0;  int        rc;  momreq = cpy_stage(momreq, pjob, JOB_ATR_stagein, STAGE_DIR_IN);  if (momreq == NULL)    {    /* no files to stage, go directly to sending job to mom */    return(svr_strtjob2(pjob, preq));    }  /* have files to stage in */  /* save job id for post_stagein */  momreq->rq_extra = malloc(PBS_MAXSVRJOBID + 1);  if (momreq->rq_extra == 0)    {    return(PBSE_SYSTEM);    }  strcpy(momreq->rq_extra, pjob->ji_qs.ji_jobid);  rc = relay_to_mom(         pjob->ji_qs.ji_un.ji_exect.ji_momaddr,         momreq,         post_stagein);  if (rc == 0)    {    svr_setjobstate(pjob, state, substate);    /*     * stage-in started ok - reply to client as copy may     * take too long to wait.     */    if (preq != NULL)      reply_ack(preq);    }  else    {    free(momreq->rq_extra);    }  return(rc);  }  /* END svr_stagein() */

void hold_job(  attribute *temphold, /* I */  void      *j)        /* I */  {  long *hold_val;  long old_hold;  int newstate;  int newsub;  attribute *pattr;  job *pjob = (job *)j;  if (pjob == NULL)    return;  hold_val = &pjob->ji_wattr[JOB_ATR_hold].at_val.at_long;  old_hold = *hold_val;  *hold_val |= temphold->at_val.at_long;  pjob->ji_wattr[JOB_ATR_hold].at_flags |= ATR_VFLAG_SET;    pattr = &pjob->ji_wattr[JOB_ATR_checkpoint];    if ((pjob->ji_qs.ji_state == JOB_STATE_RUNNING) &&      ((pattr->at_flags & ATR_VFLAG_SET) &&       ((csv_find_string(pattr->at_val.at_str, "s") != NULL) ||        (csv_find_string(pattr->at_val.at_str, "c") != NULL) ||        (csv_find_string(pattr->at_val.at_str, "enabled") != NULL))))    {    /* TODO */    /* preq_tmp = alloc_br(preq->rq_type); */        }  else if (old_hold != *hold_val)    {    /* indicate attributes changed  */        pjob->ji_modified = 1;    svr_evaljobstate(pjob, &newstate, &newsub, 0);    svr_setjobstate(pjob, newstate, newsub);    }  }

/** * @brief * 		force_reque - requeue (rerun) a job * * @param[in,out]	pwt	-	job which needs to be rerun */voidforce_reque(job *pjob){	int  newstate;	int  newsubstate;	pjob->ji_modified = 1;	pjob->ji_momhandle = -1;	pjob->ji_mom_prot = PROT_INVALID;	/* simulate rerun: free nodes, clear checkpoint flag, and */	/* clear exec_vnode string				  */	rel_resc(pjob);	/* note in accounting file */	account_jobend(pjob, pjob->ji_acctrec, PBS_ACCT_RERUN);	/* if a subjob,  we set substate to RERUN3 to cause trktbl entry */	/* to be reset to Qeued, and then blow away the job struct       */	if (pjob->ji_qs.ji_svrflags & JOB_SVFLG_SubJob) {		pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN3;		job_purge(pjob);		return;	}	/*	 * Clear any JOB_SVFLG_Actsuspd flag too, as the job is no longer	 * suspended (User busy).  A suspended job is rerun in case of a	 * MOM failure after the workstation becomes active(busy).	 */	pjob->ji_qs.ji_svrflags &= ~(JOB_SVFLG_Actsuspd | JOB_SVFLG_StagedIn | JOB_SVFLG_CHKPT);	job_attr_def[(int)JOB_ATR_exec_host].at_free(		&pjob->ji_wattr[(int)JOB_ATR_exec_host]);	job_attr_def[(int)JOB_ATR_exec_host2].at_free(		&pjob->ji_wattr[(int)JOB_ATR_exec_host2]);	job_attr_def[(int)JOB_ATR_exec_vnode].at_free(		&pjob->ji_wattr[(int)JOB_ATR_exec_vnode]);	job_attr_def[(int)JOB_ATR_pset].at_free(		&pjob->ji_wattr[(int)JOB_ATR_pset]);	/* job dir has no meaning for re-queued jobs, so unset it */	job_attr_def[(int)JOB_ATR_jobdir].at_free(&pjob->		ji_wattr[(int)JOB_ATR_jobdir]);	svr_evaljobstate(pjob, &newstate, &newsubstate, 1);	(void)svr_setjobstate(pjob, newstate, newsubstate);}

intshutdown_preempt_chkpt(job *pjob){	struct batch_request *phold;	attribute temp;	void (*func)(struct work_task *);	long *hold_val = NULL;	long old_hold = 0;	phold = alloc_br(PBS_BATCH_HoldJob);	if (phold == NULL)		return (PBSE_SYSTEM);	temp.at_flags = ATR_VFLAG_SET;	temp.at_type  = job_attr_def[(int)JOB_ATR_hold].at_type;	temp.at_user_encoded = NULL;	temp.at_priv_encoded = NULL;	temp.at_val.at_long = HOLD_s;	phold->rq_perm = ATR_DFLAG_MGRD | ATR_DFLAG_MGWR;	(void)strcpy(phold->rq_ind.rq_hold.rq_orig.rq_objname, pjob->ji_qs.ji_jobid);	CLEAR_HEAD(phold->rq_ind.rq_hold.rq_orig.rq_attr);	if (job_attr_def[(int)JOB_ATR_hold].at_encode(&temp,		&phold->rq_ind.rq_hold.rq_orig.rq_attr,		job_attr_def[(int)JOB_ATR_hold].at_name,		NULL,		ATR_ENCODE_CLIENT, NULL) < 0)		return (PBSE_SYSTEM);	phold->rq_extra = pjob;	func = post_chkpt;	if (relay_to_mom(pjob, phold, func) == 0) {		if (pjob->ji_qs.ji_state == JOB_STATE_TRANSIT)			svr_setjobstate(pjob, JOB_STATE_RUNNING, JOB_SUBSTATE_RUNNING);		pjob->ji_qs.ji_svrflags |= (JOB_SVFLG_HASRUN | JOB_SVFLG_CHKPT | JOB_SVFLG_HASHOLD);		pjob->ji_modified = 1;		(void)job_save(pjob, SAVEJOB_QUICK);		return (0);	} else {		*hold_val = old_hold;	/* reset to the old value */		return (-1);	}}

END_TESTSTART_TEST(svr_setjobstate_test)  {  struct job test_job;  int result = PBSE_NONE;  memset(&test_job, 0, sizeof(test_job));  /*initialize_globals*/  server.sv_qs_mutex = (pthread_mutex_t *)calloc(1, sizeof(pthread_mutex_t));  server.sv_attr_mutex = (pthread_mutex_t *)calloc(1, sizeof(pthread_mutex_t));  server.sv_jobstates_mutex = (pthread_mutex_t *)calloc(1, sizeof(pthread_mutex_t));  pthread_mutex_init(server.sv_qs_mutex,NULL);  pthread_mutex_init(server.sv_attr_mutex,NULL);  pthread_mutex_init(server.sv_jobstates_mutex,NULL);  result = svr_setjobstate(NULL, 0, 0, 0);  fail_unless(result == PBSE_BAD_PARAMETER, "NULL input pointer fail");  result = svr_setjobstate(&test_job, 0, 0, 0);  fail_unless(result == PBSE_NONE, "svr_setjobstate fail");  result = svr_setjobstate(&test_job, 1, 2, 3);  fail_unless(result == PBSE_NONE, "svr_setjobstate fail");  test_job.ji_qs.ji_state = JOB_STATE_RUNNING;  test_job.ji_wattr[JOB_ATR_exec_host].at_val.at_str = strdup("napali/0");  fail_unless(svr_setjobstate(&test_job, JOB_STATE_QUEUED, JOB_SUBSTATE_QUEUED, FALSE) == PBSE_NONE);  fail_unless(test_job.ji_wattr[JOB_ATR_exec_host].at_val.at_str == NULL);  test_job.ji_qs.ji_state = JOB_STATE_RUNNING;  test_job.ji_wattr[JOB_ATR_exec_host].at_val.at_str = strdup("lei/0");  test_job.ji_wattr[JOB_ATR_checkpoint].at_val.at_str = strdup("enabled");  test_job.ji_qs.ji_svrflags |= JOB_SVFLG_CHECKPOINT_FILE;  fail_unless(svr_setjobstate(&test_job, JOB_STATE_QUEUED, JOB_SUBSTATE_QUEUED, FALSE) == PBSE_NONE);  fail_unless(test_job.ji_wattr[JOB_ATR_exec_host].at_val.at_str != NULL, "exec_host list got removed when it shouldn't have...");  decrement_count = 0;  fail_unless(svr_setjobstate(&test_job, JOB_STATE_COMPLETE, JOB_SUBSTATE_COMPLETE, FALSE) == PBSE_NONE);  fail_unless(decrement_count == 2);  decrement_count = 0;  fail_unless(svr_setjobstate(&test_job, JOB_STATE_COMPLETE, JOB_SUBSTATE_COMPLETE, FALSE) == PBSE_NONE);  fail_unless(decrement_count == 0);  }

/* * requeue_job_without_contacting_mom() * */void requeue_job_without_contacting_mom(  job &pjob)  {  if (pjob.ji_qs.ji_state == JOB_STATE_RUNNING)    {    rel_resc(&pjob);    svr_setjobstate(&pjob, JOB_STATE_QUEUED, JOB_SUBSTATE_QUEUED, FALSE);    pjob.ji_wattr[JOB_ATR_exec_host].at_flags &= ~ATR_VFLAG_SET;    if (pjob.ji_wattr[JOB_ATR_exec_host].at_val.at_str != NULL)      {      free(pjob.ji_wattr[JOB_ATR_exec_host].at_val.at_str);      pjob.ji_wattr[JOB_ATR_exec_host].at_val.at_str = NULL;      }    }  } /* END requeue_job_without_contacting_mom() */

void force_purge_work(  job *pjob)  {  char       log_buf[LOCAL_LOG_BUF_SIZE];  pbs_queue *pque;  snprintf(log_buf, sizeof(log_buf), "purging job %s without checking MOM", pjob->ji_qs.ji_jobid);  log_event(PBSEVENT_JOB,PBS_EVENTCLASS_JOB,pjob->ji_qs.ji_jobid,log_buf);    free_nodes(pjob);  if ((pque = get_jobs_queue(&pjob)) != NULL)    {    mutex_mgr pque_mutex = mutex_mgr(pque->qu_mutex, true);    if (pque->qu_qs.qu_type == QTYPE_Execution)      {      pque_mutex.unlock();      set_resc_assigned(pjob, DECR);      }    }  depend_on_term(pjob);  svr_setjobstate(pjob, JOB_STATE_COMPLETE, JOB_SUBSTATE_COMPLETE, FALSE);    if (pjob != NULL)    {    if (is_ms_on_server(pjob))      {      char  log_buf[LOCAL_LOG_BUF_SIZE];      if (LOGLEVEL >= 7)        {        snprintf(log_buf, sizeof(log_buf), "Mother Superior is on the server, not cleaning spool files in %s", __func__);        log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buf);        }      svr_job_purge(pjob, 1);      }    else      svr_job_purge(pjob);    }  } /* END force_purge_work() */

static intshutdown_chkpt(job *pjob){	struct batch_request *phold;	attribute 	      temp;	phold = alloc_br(PBS_BATCH_HoldJob);	if (phold == (struct batch_request *)0)		return (PBSE_SYSTEM);	temp.at_flags = ATR_VFLAG_SET;	temp.at_type  = job_attr_def[(int)JOB_ATR_hold].at_type;	temp.at_user_encoded = NULL;	temp.at_priv_encoded = NULL;	temp.at_val.at_long = HOLD_s;	phold->rq_perm = ATR_DFLAG_MGRD | ATR_DFLAG_MGWR;	(void)strcpy(phold->rq_ind.rq_hold.rq_orig.rq_objname, pjob->ji_qs.ji_jobid);	CLEAR_HEAD(phold->rq_ind.rq_hold.rq_orig.rq_attr);	if (job_attr_def[(int)JOB_ATR_hold].at_encode(&temp,		&phold->rq_ind.rq_hold.rq_orig.rq_attr,		job_attr_def[(int)JOB_ATR_hold].at_name,		(char *)0,		ATR_ENCODE_CLIENT, NULL) < 0)		return (PBSE_SYSTEM);	if (relay_to_mom(pjob, phold, post_chkpt) == 0) {		if (pjob->ji_qs.ji_state == JOB_STATE_TRANSIT)			svr_setjobstate(pjob, JOB_STATE_RUNNING, JOB_SUBSTATE_RUNNING);		pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN;		pjob->ji_qs.ji_svrflags |= JOB_SVFLG_HASRUN;		pjob->ji_qs.ji_svrflags |= JOB_SVFLG_CHKPT;		pjob->ji_modified = 1;		(void)job_save(pjob, SAVEJOB_QUICK);		return (0);	} else		return (-1);}

intnet_move(job *jobp, struct batch_request *req){	void		*data;	char		*destination = jobp->ji_qs.ji_destin;	pbs_net_t	 hostaddr;	char		*hostname;	int		 move_type;	unsigned int	 port = pbs_server_port_dis;	void	       (*post_func)(struct work_task *);	char		*toserver;	/* Determine to whom are we sending the job */	if ((toserver = strchr(destination, '@')) == NULL) {		sprintf(log_buffer,			"no server specified in %s", destination);		log_err(-1, __func__, log_buffer);		return (-1);	}	toserver++;		/* point to server name */	hostname = parse_servername(toserver, &port);	hostaddr = get_hostaddr(hostname);	if (req) {		/* note, in this case, req is the orginal Move Request */		move_type = MOVE_TYPE_Move;		post_func = post_movejob;		data      = req;	} else {		/* note, in this case req is NULL */		move_type = MOVE_TYPE_Route;		post_func = post_routejob;		data      = 0;	}	(void)svr_setjobstate(jobp, JOB_STATE_TRANSIT, JOB_SUBSTATE_TRNOUT);	return (send_job(jobp, hostaddr, port, move_type, post_func, data));}

static void post_routejob(  struct work_task *pwt)  {  int  newstate;  int  newsub;  int  r;  int  stat = pwt->wt_aux;  char *id = "post_routejob";  job *jobp = (job *)pwt->wt_parm1;  if (WIFEXITED(stat))    {    r = WEXITSTATUS(stat);    }  else    {    r = 2;    sprintf(log_buffer, msg_badexit,            stat);    strcat(log_buffer, id);    log_event(      PBSEVENT_SYSTEM,      PBS_EVENTCLASS_JOB,      jobp->ji_qs.ji_jobid,      log_buffer);    }  switch (r)    {    case 0:  /* normal return, job was routed */      if (jobp->ji_qs.ji_svrflags & JOB_SVFLG_StagedIn)        remove_stagein(jobp);      if (jobp->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_COPIED)        remove_checkpoint(jobp);      job_purge(jobp); /* need to remove server job struct */      return;      /*NOTREACHED*/      break;    case 1:  /* permanent rejection (or signal) */      if (jobp->ji_qs.ji_substate == JOB_SUBSTATE_ABORT)        {        /* job delete in progress, just set to queued status */        svr_setjobstate(jobp, JOB_STATE_QUEUED, JOB_SUBSTATE_ABORT);        return;        }      add_dest(jobp);  /* else mark destination as bad */      /* fall through */    default : /* try routing again */      /* force re-eval of job state out of Transit */      svr_evaljobstate(jobp, &newstate, &newsub, 1);      svr_setjobstate(jobp, newstate, newsub);      if ((r = job_route(jobp)) == PBSE_ROUTEREJ)        job_abt(&jobp, pbse_to_txt(PBSE_ROUTEREJ));      else if (r != 0)        job_abt(&jobp, msg_routexceed);      break;    }  /* END switch (r) */  return;  }  /* END post_routejob() */

/** * attempt_delete() * deletes a job differently depending on the job's state * * @return TRUE if the job was deleted, FALSE if skipped * @param pjob - a pointer to the job being handled */int attempt_delete(  void *j) /* I */  {  int skipped = FALSE;  struct work_task *pwtold;  struct work_task *pwtnew;  job *pjob;  /* job considered deleted if null */  if (j == NULL)    return(TRUE);  pjob = (job *)j;  if (pjob->ji_qs.ji_state == JOB_STATE_TRANSIT)    {    /*     * Find pid of router from existing work task entry,     * then establish another work task on same child.     * Next, signal the router and wait for its completion;     */        pwtold = (struct work_task *)GET_NEXT(pjob->ji_svrtask);        while (pwtold != NULL)      {      if ((pwtold->wt_type == WORK_Deferred_Child) ||          (pwtold->wt_type == WORK_Deferred_Cmp))        {        kill((pid_t)pwtold->wt_event, SIGTERM);                pjob->ji_qs.ji_substate = JOB_SUBSTATE_ABORT;        }            pwtold = (struct work_task *)GET_NEXT(pwtold->wt_linkobj);      }    skipped = TRUE;        return(!skipped);    }  /* END if (pjob->ji_qs.ji_state == JOB_SUBSTATE_TRANSIT) */  else if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)    {    /* we'll wait for the mom to get this job, then delete it */    skipped = TRUE;    }  /* END if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN) */  else if (pjob->ji_qs.ji_state == JOB_STATE_RUNNING)    {    /* set up nanny */        if (!has_job_delete_nanny(pjob))      {      apply_job_delete_nanny(pjob, time_now + 60);            /* need to issue a signal to the mom, but we don't want to sent an ack to the       * client when the mom replies */      issue_signal(pjob, "SIGTERM", post_delete, NULL);      }    if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_FILE) != 0)      {      /* job has restart file at mom, change restart comment if failed */      change_restart_comment_if_needed(pjob);      }        return(!skipped);    }  /* END if (pjob->ji_qs.ji_state == JOB_STATE_RUNNING) */  if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_FILE) != 0)    {    /* job has restart file at mom, change restart comment if failed */        change_restart_comment_if_needed(pjob);        /* job has restart file at mom, do end job processing */    svr_setjobstate(pjob, JOB_STATE_EXITING, JOB_SUBSTATE_EXITING);    pjob->ji_momhandle = -1;        /* force new connection */    pwtnew = set_task(WORK_Immed, 0, on_job_exit, (void *)pjob);        if (pwtnew)      {      append_link(&pjob->ji_svrtask, &pwtnew->wt_linkobj, pwtnew);      }       }  else if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_StagedIn) != 0)    {//.........这里部分代码省略.........

void array_delete_wt(struct work_task *ptask)  {  struct batch_request *preq;  job_array *pa;  /*struct work_task *pnew_task;*/  struct work_task *pwtnew;  int i;  static int last_check = 0;  static char *last_id = NULL;  preq = ptask->wt_parm1;  pa = get_array(preq->rq_ind.rq_delete.rq_objname);  if (pa == NULL)    {    /* jobs must have exited already */    reply_ack(preq);    last_check = 0;    free(last_id);    last_id = NULL;    return;    }  if (last_id == NULL)    {    last_id = strdup(preq->rq_ind.rq_delete.rq_objname);    last_check = time_now;    }  else if (strcmp(last_id, preq->rq_ind.rq_delete.rq_objname) != 0)    {    last_check = time_now;    free(last_id);    last_id = strdup(preq->rq_ind.rq_delete.rq_objname);    }  else if (time_now - last_check > 10)    {    int num_jobs;    int num_prerun;    job *pjob;    num_jobs = 0;    num_prerun = 0;    for (i = 0; i < pa->ai_qs.array_size; i++)      {      if (pa->jobs[i] == NULL)        continue;      pjob = (job *)pa->jobs[i];      num_jobs++;      if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)        {        num_prerun++;        /* mom still hasn't gotten job?? delete anyway */        if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_FILE) != 0)          {          /* job has restart file at mom, do end job processing */          change_restart_comment_if_needed(pjob);          svr_setjobstate(pjob, JOB_STATE_EXITING, JOB_SUBSTATE_EXITING);          pjob->ji_momhandle = -1;          /* force new connection */          pwtnew = set_task(WORK_Immed, 0, on_job_exit, (void *)pjob);          if (pwtnew)            {            append_link(&pjob->ji_svrtask, &pwtnew->wt_linkobj, pwtnew);            }          }        else if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_StagedIn) != 0)          {          /* job has staged-in file, should remove them */          remove_stagein(pjob);          job_abt(&pjob, NULL);          }        else          {          job_abt(&pjob, NULL);          }        }      }    if (num_jobs == num_prerun)//.........这里部分代码省略.........

void stat_update(      struct batch_request *preq,  struct stat_cntl     *cntl)  {  job                  *pjob;  struct batch_reply   *preply;  struct brp_status    *pstatus;  svrattrl             *sattrl;  int                   oldsid;  int                   bad = 0;  time_t                time_now = time(NULL);  char                 *msg_ptr = NULL;  char                  log_buf[LOCAL_LOG_BUF_SIZE];  preply = &preq->rq_reply;  if (preply->brp_un.brp_txt.brp_str != NULL)    {    msg_ptr = strstr(preply->brp_un.brp_txt.brp_str, PBS_MSG_EQUAL);      if (msg_ptr != NULL)      msg_ptr += strlen(PBS_MSG_EQUAL);    }  if (preply->brp_choice == BATCH_REPLY_CHOICE_Status)    {    pstatus = (struct brp_status *)GET_NEXT(preply->brp_un.brp_status);    while (pstatus != NULL)      {      if ((pjob = svr_find_job(pstatus->brp_objname, FALSE)) != NULL)        {        mutex_mgr job_mutex(pjob->ji_mutex, true);        sattrl = (svrattrl *)GET_NEXT(pstatus->brp_attr);        oldsid = pjob->ji_wattr[JOB_ATR_session_id].at_val.at_long;        modify_job_attr(          pjob,          sattrl,          ATR_DFLAG_MGWR | ATR_DFLAG_SvWR,          &bad);        if (oldsid != pjob->ji_wattr[JOB_ATR_session_id].at_val.at_long)          {          /* first save since running job (or the sid has changed), */          /* must save session id    */          job_save(pjob, SAVEJOB_FULL, 0);          }#ifdef USESAVEDRESOURCES        else          {          /* save so we can recover resources used */          job_save(pjob, SAVEJOB_FULL, 0);          }#endif    /* USESAVEDRESOURCES */        pjob->ji_momstat = time_now;        }      pstatus = (struct brp_status *)GET_NEXT(pstatus->brp_stlink);      }  /* END while (pstatus != NULL) */    }    /* END if (preply->brp_choice == BATCH_REPLY_CHOICE_Status) */  else if ((preply->brp_choice == BATCH_REPLY_CHOICE_Text) &&           (preply->brp_code == PBSE_UNKJOBID) &&           (msg_ptr != NULL) &&           (!strcmp(msg_ptr,  preq->rq_ind.rq_status.rq_id)))    {    /* we sent a stat request, but mom says it doesn't know anything about       the job */    if ((pjob = svr_find_job(preq->rq_ind.rq_status.rq_id, FALSE)) != NULL)      {      /* job really isn't running any more - mom doesn't know anything about it         this can happen if a diskless node reboots and the mom_priv/jobs         directory is cleared, set its state to queued so job_abt doesn't         think it is still running */      mutex_mgr job_mutex(pjob->ji_mutex, true);            snprintf(log_buf, sizeof(log_buf),        "mother superior no longer recognizes %s as a valid job, aborting. Last reported time was %ld",        preq->rq_ind.rq_status.rq_id, pjob->ji_last_reported_time);      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buf);            svr_setjobstate(pjob, JOB_STATE_QUEUED, JOB_SUBSTATE_ABORT, FALSE);      rel_resc(pjob);      job_mutex.set_unlock_on_exit(false);      job_abt(&pjob, "Job does not exist on node");      /* TODO, if the job is rerunnable we should set its state back to queued */      }    }  else    {    snprintf(log_buf, sizeof(log_buf),      "Poll job request failed for job %s", preq->rq_ind.rq_status.rq_id);//.........这里部分代码省略.........

void process_hold_reply(  batch_request *preq)  {  job                  *pjob;  pbs_attribute         temphold;  int                   newstate;  int                   newsub;  int                   rc;  char                 *pset;  char                  log_buf[LOCAL_LOG_BUF_SIZE];  /* preq was handled previously */  if (preq == NULL)    return;  preq->rq_conn = preq->rq_orgconn;  /* restore client socket */  if ((pjob = svr_find_job(preq->rq_ind.rq_hold.rq_orig.rq_objname, FALSE)) == NULL)    {    log_event(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB,              preq->rq_ind.rq_hold.rq_orig.rq_objname,              msg_postmomnojob);    req_reject(PBSE_UNKJOBID, 0, preq, NULL, msg_postmomnojob);    }  else    {    mutex_mgr job_mutex(pjob->ji_mutex, true);    if (preq->rq_reply.brp_code != 0)      {      rc = get_hold(&preq->rq_ind.rq_hold.rq_orig.rq_attr, (const char **)&pset, &temphold);            if (rc == 0)        {        rc = job_attr_def[JOB_ATR_hold].at_set(&pjob->ji_wattr[JOB_ATR_hold],            &temphold, DECR);        }            pjob->ji_qs.ji_substate = JOB_SUBSTATE_RUNNING;  /* reset it */            pjob->ji_modified = 1;    /* indicate attributes changed */      svr_evaljobstate(*pjob, newstate, newsub, 0);      svr_setjobstate(pjob, newstate, newsub, FALSE); /* saves job */            if (preq->rq_reply.brp_code != PBSE_NOSUP)        {        sprintf(log_buf, msg_mombadhold, preq->rq_reply.brp_code);        log_event(PBSEVENT_DEBUG, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buf);        req_reject(preq->rq_reply.brp_code, 0, preq, NULL, log_buf);        }      else        {        reply_ack(preq);        }      }    else      {      /* record that MOM has a checkpoint file */            /* PBS_CHECKPOINT_MIGRATEABLE is defined as zero therefore this code will never fire.       * And if these flags are not set, start_exec will not try to run the job from       * the checkpoint image file.       */      pjob->ji_qs.ji_svrflags |= JOB_SVFLG_CHECKPOINT_FILE;            if (preq->rq_reply.brp_auxcode)  /* checkpoint can be moved */        {        pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_CHECKPOINT_FILE;        pjob->ji_qs.ji_svrflags |=  JOB_SVFLG_HASRUN | JOB_SVFLG_CHECKPOINT_MIGRATEABLE;        }      pjob->ji_modified = 1;    /* indicate attributes changed     */            svr_evaljobstate(*pjob, newstate, newsub, 0);      svr_setjobstate(pjob, newstate, newsub, FALSE); /* saves job */            account_record(PBS_ACCT_CHKPNT, pjob, "Checkpointed and held"); /* note in accounting file */      reply_ack(preq);      }    }  } /* END process_hold_reply() */

/** * attempt_delete() * deletes a job differently depending on the job's state * * @return TRUE if the job was deleted, FALSE if skipped * @param pjob - a pointer to the job being handled */int attempt_delete(  void *j) /* I */  {  int        skipped = FALSE;  int        release_mutex = TRUE;  job       *pjob;  time_t     time_now = time(NULL);  char       log_buf[LOCAL_LOG_BUF_SIZE];  /* job considered deleted if null */  if (j == NULL)    return(TRUE);  pjob = (job *)j;  if (pjob->ji_qs.ji_state == JOB_STATE_TRANSIT)    {    /* I'm not sure if this is still possible since the thread     * waits on the job to finish transmiting, but I'll leave     * this part here --dbeer */    skipped = TRUE;        return(!skipped);    }  /* END if (pjob->ji_qs.ji_state == JOB_SUBSTATE_TRANSIT) */  else if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)    {    /* we'll wait for the mom to get this job, then delete it */    skipped = TRUE;    }  /* END if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN) */  else if (pjob->ji_qs.ji_state == JOB_STATE_RUNNING)    {    /* set up nanny */        if (pjob->ji_has_delete_nanny == FALSE)      {      apply_job_delete_nanny(pjob, time_now + 60);            /* need to issue a signal to the mom, but we don't want to sent an ack to the       * client when the mom replies */      issue_signal(&pjob, "SIGTERM", post_delete, NULL);      }    if (pjob != NULL)      {      if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_FILE) != 0)        {        /* job has restart file at mom, change restart comment if failed */        change_restart_comment_if_needed(pjob);        }      unlock_ji_mutex(pjob, __func__, "1", LOGLEVEL);      }        return(!skipped);    }  /* END if (pjob->ji_qs.ji_state == JOB_STATE_RUNNING) */  if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_FILE) != 0)    {    /* job has restart file at mom, change restart comment if failed */        change_restart_comment_if_needed(pjob);        /* job has restart file at mom, do end job processing */    svr_setjobstate(pjob, JOB_STATE_EXITING, JOB_SUBSTATE_EXITING, FALSE);    pjob->ji_momhandle = -1;        /* force new connection */    if (LOGLEVEL >= 7)      {      sprintf(log_buf, "calling on_job_exit from %s", __func__);      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buf);      }    set_task(WORK_Immed, 0, on_job_exit, strdup(pjob->ji_qs.ji_jobid), FALSE);    }  else if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_StagedIn) != 0)    {    /* job has staged-in file, should remove them */        remove_stagein(&pjob);        if (pjob != NULL)      job_abt(&pjob, NULL);    release_mutex = FALSE;    }  else    {//.........这里部分代码省略.........

void array_delete_wt(      struct work_task *ptask)  {  struct batch_request *preq;  job_array            *pa;  int                   i;  char                  log_buf[LOCAL_LOG_BUF_SIZE];  int                   num_jobs = 0;  int                   num_prerun = 0;  job                  *pjob;  preq = get_remove_batch_request((char *)ptask->wt_parm1);    free(ptask->wt_mutex);  free(ptask);  if (preq == NULL)    return;  pa = get_array(preq->rq_ind.rq_delete.rq_objname);  if (pa == NULL)    {    /* jobs must have exited already */    reply_ack(preq);    return;    }  for (i = 0; i < pa->ai_qs.array_size; i++)    {    if (pa->job_ids[i] == NULL)      continue;        if ((pjob = svr_find_job(pa->job_ids[i], FALSE)) == NULL)      {      free(pa->job_ids[i]);      pa->job_ids[i] = NULL;      }    else      {      num_jobs++;            if (pjob->ji_qs.ji_substate == JOB_SUBSTATE_PRERUN)        {        num_prerun++;        /* mom still hasn't gotten job?? delete anyway */                if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_FILE) != 0)          {          /* job has restart file at mom, do end job processing */          change_restart_comment_if_needed(pjob);                    svr_setjobstate(pjob, JOB_STATE_EXITING, JOB_SUBSTATE_EXITING, FALSE);                    pjob->ji_momhandle = -1;                    /* force new connection */          if (LOGLEVEL >= 7)            {            sprintf(log_buf, "calling on_job_exit from %s", __func__);            log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buf);            }          set_task(WORK_Immed, 0, on_job_exit, strdup(pjob->ji_qs.ji_jobid), FALSE);                    unlock_ji_mutex(pjob, __func__, "1", LOGLEVEL);          }        }      else if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_StagedIn) != 0)        {        /* job has staged-in file, should remove them */        remove_stagein(&pjob);                if (pjob != NULL)          {          /* job_abt() calls svr_job_purge which will try to lock the array again */          pthread_mutex_unlock(pa->ai_mutex);          job_abt(&pjob, NULL);          pthread_mutex_lock(pa->ai_mutex);          }        }      else        {        /* job_abt() calls svr_job_purge which will try to lock the array again */        pthread_mutex_unlock(pa->ai_mutex);        job_abt(&pjob, NULL);        pthread_mutex_lock(pa->ai_mutex);        }      } /* END if (ji_substate == JOB_SUBSTATE_PRERUN) */    } /* END for each job in array */    pthread_mutex_unlock(pa->ai_mutex);  if (LOGLEVEL >= 7)    {    sprintf(log_buf, "%s: unlocked ai_mutex", __func__);    log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, __func__, log_buf);//.........这里部分代码省略.........

/** * update_array_values() * * updates internal bookeeping values for job arrays * @param pa - array to update * @param pjob - the pjob that an event happened on * @param event - code for what event just happened */void update_array_values(  job_array            *pa,        /* I */  void                 *j,         /* I */  int                   old_state, /* I */  enum ArrayEventsEnum  event)     /* I */  {  job *pjob = (job *)j;  int exit_status;  switch (event)    {    case aeQueue:      /* NYI, nothing needs to be done for this yet */      break;    case aeRun:      if (old_state != JOB_STATE_RUNNING)        {        pa->ai_qs.jobs_running++;        pa->ai_qs.num_started++;        }      break;    case aeTerminate:      exit_status = pjob->ji_qs.ji_un.ji_exect.ji_exitstat;      if (old_state == JOB_STATE_RUNNING)        {        if (pa->ai_qs.jobs_running > 0)          pa->ai_qs.jobs_running--;        }      if (exit_status == 0)        {        pa->ai_qs.num_successful++;        pa->ai_qs.jobs_done++;        }      else        {        pa->ai_qs.num_failed++;        pa->ai_qs.jobs_done++;        }      array_save(pa);      /* update slot limit hold if necessary */      if (server.sv_attr[SRV_ATR_MoabArrayCompatible].at_val.at_long != FALSE)        {        /* only need to update if the job wasn't previously held */        if ((pjob->ji_wattr[JOB_ATR_hold].at_val.at_long & HOLD_l) == FALSE)          {          int  i;          int  newstate;          int  newsub;          job *pj;          /* find the first held job and release its hold */          for (i = 0; i < pa->ai_qs.array_size; i++)            {            if (pa->jobs[i] == NULL)              continue;            pj = (job *)pa->jobs[i];            if (pj->ji_wattr[JOB_ATR_hold].at_val.at_long & HOLD_l)              {              pj->ji_wattr[JOB_ATR_hold].at_val.at_long &= ~HOLD_l;              if (pj->ji_wattr[JOB_ATR_hold].at_val.at_long == 0)                {                pj->ji_wattr[JOB_ATR_hold].at_flags &= ~ATR_VFLAG_SET;                }                           svr_evaljobstate(pj, &newstate, &newsub, 1);              svr_setjobstate(pj, newstate, newsub);              job_save(pj, SAVEJOB_FULL, 0);              break;              }            }          }        }      break;    default://.........这里部分代码省略.........

static void post_movejob(  struct work_task *pwt)  {  char *id = "post_movejob";  struct batch_request *req;  int newstate;  int newsub;  int stat;  int r;  job *jobp;  req  = (struct batch_request *)pwt->wt_parm2;  stat = pwt->wt_aux;  pbs_errno = PBSE_NONE;  if (req->rq_type != PBS_BATCH_MoveJob)    {    sprintf(log_buffer, "bad request type %d/n",            req->rq_type);    log_err(-1, id, log_buffer);    return;    }  jobp = find_job(req->rq_ind.rq_move.rq_jid);  if ((jobp == NULL) || (jobp != (job *)pwt->wt_parm1))    {    sprintf(log_buffer, "job %s not found/n",            req->rq_ind.rq_move.rq_jid);    log_err(-1, id, log_buffer);    }  if (WIFEXITED(stat))    {    r = WEXITSTATUS(stat);    if (r == 0)      {      /* purge server's job structure */      if (jobp->ji_qs.ji_svrflags & JOB_SVFLG_StagedIn)        remove_stagein(jobp);      if (jobp->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_COPIED)        remove_checkpoint(jobp);      strcpy(log_buffer, msg_movejob);      sprintf(log_buffer + strlen(log_buffer), msg_manager,              req->rq_ind.rq_move.rq_destin,              req->rq_user,              req->rq_host);      job_purge(jobp);      }    else      {      r = PBSE_ROUTEREJ;      }    }  else    {    r = PBSE_SYSTEM;    sprintf(log_buffer, msg_badexit, stat);    strcat(log_buffer, id);    log_event(      PBSEVENT_SYSTEM,      PBS_EVENTCLASS_JOB,      jobp->ji_qs.ji_jobid,      log_buffer);    }  if (r)    {    if (jobp != NULL)      {      /* force re-eval of job state out of Transit */      svr_evaljobstate(jobp, &newstate, &newsub, 1);      svr_setjobstate(jobp, newstate, newsub);      }    req_reject(r, 0, req, NULL, NULL);    }  else    {    reply_ack(req);    }//.........这里部分代码省略.........

//.........这里部分代码省略.........			log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_ERR,				pjob->ji_qs.ji_jobid, log_buffer);			req_reject(PBSE_PERM, 0, preq);			return;		}		plist = (svrattrl *)GET_NEXT(plist->al_link);	}	if (outsideselect) {		presc = find_resc_entry(&pjob->ji_wattr[(int)JOB_ATR_resource],			pseldef);		if (presc &&			((presc->rs_value.at_flags & ATR_VFLAG_DEFLT) == 0)) {			/* select is not a default, so reject qalter */			resc_in_err = strdup(outsideselect->rs_name);			req_reject(PBSE_INVALJOBRESC, 0, preq);			return;		}	}	/* modify the jobs attributes */	bad = 0;	plist = (svrattrl *)GET_NEXT(preq->rq_ind.rq_modify.rq_attr);	rc = modify_job_attr(pjob, plist, preq->rq_perm, &bad);	if (rc) {		if (pjob->ji_clterrmsg)			reply_text(preq, rc, pjob->ji_clterrmsg);		else			reply_badattr(rc, bad, plist, preq);		return;	}	/* If certain attributes modified and if in scheduling cycle  */	/* then add to list of jobs which cannot be run in this cycle */	if (add_to_am_list)		am_jobs_add(pjob);	/* see req_runjob() */	/* check if project attribute was requested to be modified to */	/* be the default project value */	if (mod_project && (pjob->ji_wattr[(int)JOB_ATR_project].at_flags & /							ATR_VFLAG_SET)) {		if (strcmp(pjob->ji_wattr[(int)JOB_ATR_project].at_val.at_str,			PBS_DEFAULT_PROJECT) == 0) {			sprintf(log_buffer, msg_defproject,				ATTR_project, PBS_DEFAULT_PROJECT);#ifdef NAS /* localmod 107 */			log_event(PBSEVENT_DEBUG4, PBS_EVENTCLASS_JOB, LOG_INFO,				pjob->ji_qs.ji_jobid, log_buffer);#else			log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_INFO,				pjob->ji_qs.ji_jobid, log_buffer);#endif /* localmod 107 */		}	}	if (pjob->ji_wattr[(int)JOB_ATR_resource].at_flags & ATR_VFLAG_MODIFY) {		presc = find_resc_entry(&pjob->ji_wattr[(int)JOB_ATR_resource],			pseldef);		if (presc && (presc->rs_value.at_flags & ATR_VFLAG_DEFLT)) {			/* changing Resource_List and select is a default   */			/* clear "select" so it is rebuilt inset_resc_deflt */			pseldef->rs_free(&presc->rs_value);		}	}	/* Reset any defaults resource limit which might have been unset */	if ((rc = set_resc_deflt((void *)pjob, JOB_OBJECT, NULL)) != 0) {		req_reject(rc, 0, preq);		return;	}	/* if job is not running, may need to change its state */	if (pjob->ji_qs.ji_state != JOB_STATE_RUNNING) {		svr_evaljobstate(pjob, &newstate, &newsubstate, 0);		(void)svr_setjobstate(pjob, newstate, newsubstate);	} else {		(void)job_save(pjob, SAVEJOB_FULL);	}	(void)sprintf(log_buffer, msg_manager, msg_jobmod,		preq->rq_user, preq->rq_host);	log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, LOG_INFO,		pjob->ji_qs.ji_jobid, log_buffer);	/* if a resource limit changed for a running job, send to MOM */	if (sendmom) {		rc = relay_to_mom(pjob, preq, post_modify_req);		if (rc)			req_reject(rc, 0, preq);    /* unable to get to MOM */		return;	}	reply_ack(preq);}

void update_array_statuses()  {  job_array *pa;  job *pj;  int i;  unsigned int running;  unsigned int queued;  unsigned int held;  unsigned int complete;    pa = (job_array*)GET_NEXT(svr_jobarrays);  while (pa != NULL)    {    running = 0;    queued = 0;    held = 0;    complete = 0;        for (i = 0; i < pa->ai_qs.array_size; i++)      {      pj = pa->jobs[i];            if (pj != NULL)        {        if (pj->ji_qs.ji_state == JOB_STATE_RUNNING)          {          running++;          }        else if (pj->ji_qs.ji_state == JOB_STATE_QUEUED)          {          queued++;          }        else if (pj->ji_qs.ji_state == JOB_STATE_HELD)          {          held++;          }        else if (pj->ji_qs.ji_state == JOB_STATE_COMPLETE)          {          complete++;          }        }      }        if (running > 0)      {      svr_setjobstate(pa->template_job, JOB_STATE_RUNNING, pa->template_job->ji_qs.ji_substate);      }    else if (held > 0 && queued == 0 && complete == 0)      {      svr_setjobstate(pa->template_job, JOB_STATE_HELD, pa->template_job->ji_qs.ji_substate);      }    else if (complete > 0 && queued == 0 && held == 0)      {      svr_setjobstate(pa->template_job, JOB_STATE_COMPLETE, pa->template_job->ji_qs.ji_substate);      }    else       {      /* default to just calling the array queued */      svr_setjobstate(pa->template_job, JOB_STATE_QUEUED, pa->template_job->ji_qs.ji_substate);      }          pa = (job_array*)GET_NEXT(pa->all_arrays);    }  }

//.........这里部分代码省略.........	if (changed_resc) {		account_entity_limit_usages(pjob, NULL,				&newattr[(int)JOB_ATR_resource], INCR, ETLIM_ACC_ALL_RES);		account_entity_limit_usages(pjob, pjob->ji_qhdr,				&newattr[(int)JOB_ATR_resource], INCR, ETLIM_ACC_ALL_RES);	}	/* Now copy the new values into the job attribute array for the purposes of running the action functions */	for (i = 0; i < JOB_ATR_LAST; i++) {		if (newattr[i].at_flags & ATR_VFLAG_MODIFY) {			/*			 * The function update_eligible_time() expects it is the only one setting accrue_type.			 * If we set it here, it will get confused.  There is no action function for accrue_type,			 * so pre-setting it for the action function calls isn't required.			 */			if (i == JOB_ATR_accrue_type)				continue;			job_attr_def[i].at_free(&pattr[i]);			if ((pre_copy[i].at_type == ATR_TYPE_LIST) ||				(pre_copy[i].at_type == ATR_TYPE_RESC)) {				list_move(&pre_copy[i].at_val.at_list,					  &pattr[i].at_val.at_list);			} else {				pattr[i] = pre_copy[i];			}			/* ATR_VFLAG_MODCACHE will be included if set */			pattr[i].at_flags = pre_copy[i].at_flags;		}	}	for (i = 0; i < JOB_ATR_LAST; i++) {		/* Check newattr instead of pattr for modify.  It is possible that		 * the attribute already has the modify flag before we added the new attributes to it.		 * We only want to call the action functions for attributes which are being modified by this function.		 */		if (newattr[i].at_flags & ATR_VFLAG_MODIFY) {			if ((job_attr_def[i].at_flags & ATR_DFLAG_NOSAVM) == 0)				modified = 1;	/* full save to disk for job */			if (job_attr_def[i].at_action) {				rc = job_attr_def[i].at_action(&newattr[i],					pjob, ATR_ACTION_ALTER);				if (rc) {					*bad = i;					break;				}			}		}	}	if (rc) {		attr_atomic_copy(pjob->ji_wattr, attr_save, job_attr_def, JOB_ATR_LAST);		free(pre_copy);		attr_atomic_kill(newattr, job_attr_def, JOB_ATR_LAST);		attr_atomic_kill(attr_save, job_attr_def, JOB_ATR_LAST);		return (rc);	}	/* The action functions may have modified the attributes, need to set them to newattr2 */	for (i = 0; i < JOB_ATR_LAST; i++) {		if (newattr[i].at_flags & ATR_VFLAG_MODIFY) {			job_attr_def[i].at_free(&pattr[i]);			switch (i) {				case JOB_ATR_state:					newstate = state_char2int(newattr[i].at_val.at_char);					break;				case JOB_ATR_substate:					newsubstate = newattr[i].at_val.at_long;					break;				case JOB_ATR_accrue_type:					newaccruetype = newattr[i].at_val.at_long;					break;				default:					if ((newattr[i].at_type == ATR_TYPE_LIST) ||						(newattr[i].at_type == ATR_TYPE_RESC)) {						list_move(&newattr[i].at_val.at_list,							  &pattr[i].at_val.at_list);					} else {						pattr[i] = newattr[i];					}			}			/* ATR_VFLAG_MODCACHE will be included if set */			pattr[i].at_flags = newattr[i].at_flags;		}	}	if (newstate != -1 && newsubstate != -1) {		svr_setjobstate(pjob, newstate, newsubstate);	}	if (newaccruetype != -1)		update_eligible_time(newaccruetype, pjob);	if (modified)		pjob->ji_modified = 1;	/* an attr was modified, do full save */	free(newattr);	free(pre_copy);	attr_atomic_kill(attr_save, job_attr_def, JOB_ATR_LAST);	return (0);}

int finalize_rerunjob(      batch_request *preq,  job           *pjob,  int            rc)  {  int       Force;  char      log_buf[LOCAL_LOG_BUF_SIZE];  if (pjob == NULL)    return(PBSE_BAD_PARAMETER);  mutex_mgr pjob_mutex = mutex_mgr(pjob->ji_mutex, true);  if (preq->rq_extend && !strncasecmp(preq->rq_extend, RERUNFORCE, strlen(RERUNFORCE)))    Force = 1;  else    Force = 0;  switch (rc)    {    case -1:      /* completed job was requeued */      /* clear out job completion time if there is one */      break;    case 0:      /* requeue request successful */      pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN;      break;    case PBSE_SYSTEM: /* This may not be accurate...*/      rc = PBSE_MEM_MALLOC;      snprintf(log_buf, LOCAL_LOG_BUF_SIZE, "Can not allocate memory");      req_reject(rc, 0, preq, NULL, log_buf);      return rc;      break;    default:      if (Force == 0)        {        rc = PBSE_MOMREJECT;        snprintf(log_buf, LOCAL_LOG_BUF_SIZE, "Rejected by mom");        req_reject(rc, 0, preq, NULL, log_buf);        return rc;        }      else        {        int           newstate;        int           newsubst;        unsigned int  dummy;        char         *tmp;        if ((cray_enabled == true) &&            (pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str != NULL))          tmp = parse_servername(pjob->ji_wattr[JOB_ATR_login_node_id].at_val.at_str, &dummy);        else          tmp = parse_servername(pjob->ji_wattr[JOB_ATR_exec_host].at_val.at_str, &dummy);        /* Cannot communicate with MOM, forcibly requeue job.           This is a relatively disgusting thing to do */        sprintf(log_buf, "rerun req to %s failed (rc=%d), forcibly requeueing job",          tmp, rc);        free(tmp);        log_event(          PBSEVENT_ERROR | PBSEVENT_ADMIN | PBSEVENT_JOB,          PBS_EVENTCLASS_JOB,          pjob->ji_qs.ji_jobid,          log_buf);        log_err(-1, __func__, log_buf);        strcat(log_buf, ", previous output files may be lost");        svr_mailowner(pjob, MAIL_OTHER, MAIL_FORCE, log_buf);        svr_setjobstate(pjob, JOB_STATE_EXITING, JOB_SUBSTATE_RERUN3, FALSE);        rel_resc(pjob); /* free resc assigned to job */        pjob->ji_modified = 1;    /* force full job save */        pjob->ji_momhandle = -1;        pjob->ji_qs.ji_svrflags &= ~JOB_SVFLG_StagedIn;        svr_evaljobstate(*pjob, newstate, newsubst, 0);        svr_setjobstate(pjob, newstate, newsubst, FALSE);        }//.........这里部分代码省略.........

//.........这里部分代码省略.........        }      }    else      {      /* why is the pque null. Something went wrong */      snprintf(log_buf, LOCAL_LOG_BUF_SIZE, "jobid %s returned a null queue", pjob->ji_qs.ji_jobid);      req_reject(PBSE_UNKQUE, 0, preq, NULL, log_buf);      return(PBSE_UNKQUE);      }        pjob->ji_qs.ji_substate = JOB_SUBSTATE_RERUN;    if (delay != 0)      {      static const char *rerun = "rerun";      char               *extra = strdup(rerun);      get_batch_request_id(preq);      /* If a qrerun -f is given requeue the job regardless of the outcome of issue_signal*/      if ((preq->rq_extend) &&           (!strncasecmp(preq->rq_extend, RERUNFORCE, strlen(RERUNFORCE))))        {        std::string extend = RERUNFORCE;        batch_request *dup = new batch_request(*preq);        get_batch_request_id(dup);        rc = issue_signal(&pjob, "SIGTERM", delay_and_send_sig_kill, extra, strdup(dup->rq_id.c_str()));        if (rc == PBSE_NORELYMOM)          {          dup->rq_reply.brp_code = PBSE_NORELYMOM;          pjob_mutex.unlock();          post_rerun(dup);          pjob = svr_find_job(preq->rq_ind.rq_signal.rq_jid, FALSE);          if (pjob == NULL)            {            delete dup;            return(PBSE_NONE);            }          pjob_mutex.set_lock_state(true);          rc = PBSE_NONE;          }        delete dup;        }      else        {        rc = issue_signal(&pjob, "SIGTERM", delay_and_send_sig_kill, extra, strdup(preq->rq_id.c_str()));        if (rc != PBSE_NONE)          {          /* cant send to MOM */          req_reject(rc, 0, preq, NULL, NULL);          }        return(rc);        }      }    else      {      static const char *rerun = "rerun";      char               *extra = strdup(rerun);      /* If a qrerun -f is given requeue the job regardless of the outcome of issue_signal*/      if (preq->rq_extend && !strncasecmp(preq->rq_extend, RERUNFORCE, strlen(RERUNFORCE)))        {        std::string extend = RERUNFORCE;        rc = issue_signal(&pjob, "SIGKILL", post_rerun, extra, strdup(extend.c_str()));        if (rc == PBSE_NORELYMOM)          rc = PBSE_NONE;        }      else        rc = issue_signal(&pjob, "SIGKILL", post_rerun, extra, NULL);      }    }  else    {     if (pjob->ji_wattr[JOB_ATR_hold].at_val.at_long == HOLD_n)      {      svr_setjobstate(pjob, JOB_STATE_QUEUED, JOB_SUBSTATE_QUEUED, FALSE);      }    else      {      svr_setjobstate(pjob, JOB_STATE_HELD, JOB_SUBSTATE_HELD, FALSE);      }    /* reset some job attributes */        pjob->ji_wattr[JOB_ATR_comp_time].at_flags &= ~ATR_VFLAG_SET;    pjob->ji_wattr[JOB_ATR_reported].at_flags &= ~ATR_VFLAG_SET;    set_statechar(pjob);    rc = -1;    }  /* finalize_rerunjob will return with pjob->ji_mutex unlocked */  pjob_mutex.set_unlock_on_exit(false);  return finalize_rerunjob(preq,pjob,rc);  }

void stat_update(      struct batch_request *preq,  struct stat_cntl     *cntl)  {  job                  *pjob;  struct batch_reply   *preply;  struct brp_status    *pstatus;  svrattrl             *sattrl;  int                   oldsid;  int                   bad = 0;  time_t                time_now = time(NULL);  preply = &preq->rq_reply;  if (preply->brp_choice == BATCH_REPLY_CHOICE_Status)    {    pstatus = (struct brp_status *)GET_NEXT(preply->brp_un.brp_status);    while (pstatus != NULL)      {      if ((pjob = svr_find_job(pstatus->brp_objname, FALSE)) != NULL)        {        sattrl = (svrattrl *)GET_NEXT(pstatus->brp_attr);        oldsid = pjob->ji_wattr[JOB_ATR_session_id].at_val.at_long;        modify_job_attr(          pjob,          sattrl,          ATR_DFLAG_MGWR | ATR_DFLAG_SvWR,          &bad);        if (oldsid != pjob->ji_wattr[JOB_ATR_session_id].at_val.at_long)          {          /* first save since running job (or the sid has changed), */          /* must save session id    */          job_save(pjob, SAVEJOB_FULL, 0);          }#ifdef USESAVEDRESOURCES        else          {          /* save so we can recover resources used */          job_save(pjob, SAVEJOB_FULL, 0);          }#endif    /* USESAVEDRESOURCES */        pjob->ji_momstat = time_now;        unlock_ji_mutex(pjob, __func__, "1", LOGLEVEL);        }      pstatus = (struct brp_status *)GET_NEXT(pstatus->brp_stlink);      }  /* END while (pstatus != NULL) */    }    /* END if (preply->brp_choice == BATCH_REPLY_CHOICE_Status) */  else    {    if (preply->brp_code == PBSE_UNKJOBID)      {      /* we sent a stat request, but mom says it doesn't know anything about         the job */      if ((pjob = svr_find_job(preq->rq_ind.rq_status.rq_id, FALSE)) != NULL)        {        /* job really isn't running any more - mom doesn't know anything about it           this can happen if a diskless node reboots and the mom_priv/jobs           directory is cleared, set its state to queued so job_abt doesn't           think it is still running */        svr_setjobstate(pjob, JOB_STATE_QUEUED, JOB_SUBSTATE_ABORT, FALSE);        rel_resc(pjob);        job_abt(&pjob, "Job does not exist on node");        /* TODO, if the job is rerunnable we should set its state back to queued */        }      }    }  cntl->sc_conn = -1;  /* MUTSU - Unlock job here? */  if (cntl->sc_post)    cntl->sc_post(cntl); /* continue where we left off */  /* If sc_post has a value it is:   * req_stat_job_step2   * if so, it expects cntl to be free'd after the call   */  free(cntl); /* a bit of a kludge but its saves an extra func */  return;  }  /* END stat_update() */

//.........这里部分代码省略.........      int        newstate;      int        newsub;      job       *tmp;      job_array *pa = get_jobs_array(&pjob);      if (pjob == NULL)        return(-1);      for (i = 0; i < pa->ai_qs.array_size; i++)        {        if (pa->job_ids[i] == NULL)          continue;        if (!strcmp(pa->job_ids[i], pjob->ji_qs.ji_jobid))          continue;        if ((tmp = svr_find_job(pa->job_ids[i], FALSE)) == NULL)          {          free(pa->job_ids[i]);          pa->job_ids[i] = NULL;          }        else          {          if (tmp->ji_wattr[JOB_ATR_hold].at_val.at_long & HOLD_l)            {            tmp->ji_wattr[JOB_ATR_hold].at_val.at_long &= ~HOLD_l;                        if (tmp->ji_wattr[JOB_ATR_hold].at_val.at_long == 0)              {              tmp->ji_wattr[JOB_ATR_hold].at_flags &= ~ATR_VFLAG_SET;              }                        svr_evaljobstate(tmp, &newstate, &newsub, 1);            svr_setjobstate(tmp, newstate, newsub, FALSE);            job_save(tmp, SAVEJOB_FULL, 0);            unlock_ji_mutex(tmp, __func__, "5", LOGLEVEL);                        break;            }          unlock_ji_mutex(tmp, __func__, "6", LOGLEVEL);          }        }      if (LOGLEVEL >= 7)        {        sprintf(log_buf, "%s: unlocking ai_mutex", __func__);        log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buf);        }      pthread_mutex_unlock(pa->ai_mutex);      }    } /* END MoabArrayCompatible check */  if ((pjob->ji_qs.ji_svrflags & JOB_SVFLG_CHECKPOINT_FILE) != 0)    {    /* job has restart file at mom, do end job processing */    svr_setjobstate(pjob, JOB_STATE_EXITING, JOB_SUBSTATE_EXITING, FALSE);    /* force new connection */    pjob->ji_momhandle = -1;    if (LOGLEVEL >= 7)      {      sprintf(log_buf, "calling on_job_exit from %s", __func__);      log_event(PBSEVENT_JOB, PBS_EVENTCLASS_JOB, pjob->ji_qs.ji_jobid, log_buf);

//.........这里部分代码省略.........  /* modify the job's attributes */  bad = 0;  plist = (svrattrl *)GET_NEXT(preq->rq_ind.rq_modify.rq_attr);  rc = modify_job_attr(pjob, plist, preq->rq_perm, &bad);  if (rc)    {    /* FAILURE */    snprintf(log_buf,sizeof(log_buf),      "Cannot set attributes for job '%s'/n",      pjob->ji_qs.ji_jobid);    log_err(rc, __func__, log_buf);    if (rc == PBSE_JOBNOTFOUND)      *j = NULL;    return(rc);    }  /* Reset any defaults resource limit which might have been unset */  set_resc_deflt(pjob, NULL, FALSE);  /* if job is not running, may need to change its state */  if (pjob->ji_qs.ji_state != JOB_STATE_RUNNING)    {    svr_evaljobstate(pjob, &newstate, &newsubstate, 0);    svr_setjobstate(pjob, newstate, newsubstate, FALSE);    }  else    {    job_save(pjob, SAVEJOB_FULL, 0);    }  sprintf(log_buf, msg_manager, msg_jobmod, preq->rq_user, preq->rq_host);  log_event(PBSEVENT_JOB,PBS_EVENTCLASS_JOB,pjob->ji_qs.ji_jobid,log_buf);  /* if a resource limit changed for a running job, send to MOM */  if (sendmom)    {    /* if the NO_MOM_RELAY flag is set the calling function will call       relay_to_mom so we do not need to do it here */    if (flag != NO_MOM_RELAY)      {      /* The last number is unused unless this is an array */      if ((rc = copy_batchrequest(&dup_req, preq, 0, -1)) != 0)        {        }      /* The dup_req is freed in relay_to_mom (failure)       * or in issue_Drequest (success) */      else if ((rc = relay_to_mom(&pjob, dup_req, post_modify_req)))        {        if (pjob != NULL)          {          snprintf(log_buf,sizeof(log_buf),            "Unable to relay information to mom for job '%s'/n",            pjob->ji_qs.ji_jobid);