job_info.c

openPBS的开放源代码

 *	  reqlist - the resource_req list
 *	  name - resoruce to look for
 *
 *	returns found resource_req or NULL
 *
 */
resource_req *find_resource_req( resource_req *reqlist, const char *name )
{
  resource_req *resreq;	

  resreq = reqlist;

  while(resreq != NULL && strcmp(resreq -> name, name) )
    resreq = resreq -> next;
  
  return resreq;
}

/*
 *
 *	free_resource_req_list - frees memory used by a resource_req list
 *
 *	  list - resource_req list
 *
 *	returns nothing
 */
void free_resource_req_list( resource_req *list )
{
  resource_req *resreq, *tmp;

  resreq = list;
  while( resreq != NULL )
  {
    tmp = resreq;
    resreq = resreq -> next;
    if( tmp -> name != NULL )
      free(tmp -> name);
    if( tmp -> res_str != NULL )
      free(tmp -> res_str);
    free(tmp);
  }
}

/*
 *
 *	print_job_info - print out a job_info struct
 *
 *	  jinfo - the job to print
 *	  brief - only print job name
 *
 *	returns nothing
 *
 */
void print_job_info( job_info *jinfo, char brief )
{
  resource_req *resreq;			/* used to print the resources */

  if( jinfo == NULL )
    return;
  
  if( jinfo -> name != NULL )
    printf("%sJob Name: %s\n", brief ? "      " : "", jinfo -> name);

  if( !brief )
  {
    if( jinfo -> comment != NULL )
      printf("comment: %s\n", jinfo -> comment);
    
    if( jinfo -> queue != NULL )
      printf("queue: %s\n", jinfo -> queue -> name);
    
    if( jinfo -> job_node )
      printf("node: %s\n", jinfo -> job_node -> name);
    
    if( jinfo -> account )
      printf("account: %s\n", jinfo -> account);
    
    if( jinfo -> group )
      printf("group: %s\n", jinfo -> group);
    
    printf("priority: %d\n", jinfo -> priority);
    printf("sch_priority: %d\n", jinfo -> sch_priority);
    printf("qtime: %d: %s", jinfo -> qtime, ctime(&(jinfo -> qtime)));
    

    printf("is_queued: %s\n", jinfo -> is_queued ? "TRUE" : "FALSE");
    printf("is_running: %s\n", jinfo -> is_running ? "TRUE" : "FALSE");
    printf("is_held: %s\n", jinfo -> is_held ? "TRUE" : "FALSE");
    printf("is_waiting: %s\n", jinfo -> is_waiting ? "TRUE" : "FALSE");
    printf("is_transit: %s\n", jinfo -> is_transit ? "TRUE" : "FALSE");
    printf("is_exiting: %s\n", jinfo -> is_exiting ? "TRUE" : "FALSE");

    printf("is_starving: %s\n", jinfo -> is_starving ? "TRUE" : "FALSE");
    printf("can_not_run: %s\n", jinfo -> can_not_run ? "TRUE" : "FALSE");
    printf("can_never_run: %s\n", jinfo -> can_never_run ? "TRUE" : "FALSE");

    resreq = jinfo -> resreq;

    while( resreq != NULL )
    {
      printf("resreq %s %ld\n", resreq -> name, resreq -> amount);
      resreq = resreq -> next;
    }
  }
}


/*
 *
 *	set_state - set the state flag in a job_info structure
 *			i.e. the is_* bit
 *
 *	  state - the state
 *	  jinfo - the job info structure
 *
 *	returns nothing
 *
 */
void set_state( char *state, job_info *jinfo )
{
  switch(state[0])
  {
    case 'Q':
      jinfo -> is_queued = 1;
    break;

    case 'R': 
      jinfo -> is_running = 1;
    break;

    case 'T':
      jinfo -> is_transit = 1;
    break;

    case 'H':
      jinfo -> is_held = 1;
    break;

    case 'W':
      jinfo -> is_waiting = 1;
    break;

    case 'E':
      jinfo -> is_exiting = 1;
    break;
    
    case 'S':
      jinfo -> is_suspended = 1;
    break;
  }
}

/*
 *
 *	update_job_on_run - update job information kept in a job_info 
 *				when a job is run
 *
 *	  pbs_sd - socket connection to pbs_server
 *	  jinfo - the job to update
 *
 *	returns nothing
 *
 */
void update_job_on_run( int pbs_sd, job_info *jinfo )
{
  jinfo -> is_queued = 0;
  jinfo -> is_running = 1;
}

/*
 *
 *	update_job_comment - update a jobs comment attribute
 *
 *	  pbs_sd - connection to the pbs_server
 *	  jinfo  - job to update 
 *	  comment - the comment string
 *
 *	returns 
 *	  0: comment needed updating
 *	  non-zero: the comment did not need to be updated (same as before etc)
 *
 */
int update_job_comment( int pbs_sd, job_info *jinfo, char *comment )
{
  /* the pbs_alterjob() call takes a linked list of attrl structures to alter
   * a job.  All we are interested in doing is changing the comment.
   */
  struct attrl attr = { NULL, ATTR_comment, NULL, NULL };

  if( jinfo == NULL )
    return 1;

  /* no need to update the job comment if it is the same */
  if( jinfo -> comment == NULL || strcmp(jinfo -> comment, comment) )
  {
    if( jinfo -> comment != NULL )
      free(jinfo -> comment);
    jinfo -> comment = string_dup( comment );
    attr.value = comment;
    pbs_alterjob(pbs_sd, jinfo -> name, &attr, NULL);
    return 0;
  }
  return 1;
}

/*
 *
 *	update_jobs_cant_run - update an array of jobs which can not run
 *
 *	  pbs_sd - connection to the PBS server
 *	  jinfo_arr - the array to update
 *	  start - the job which couldn't run
 *	  comment - the comment to update
 *
 *	returns nothing;
 *
 */
void update_jobs_cant_run(int pbs_sd, job_info **jinfo_arr, job_info *start, 
				char *comment, int start_where)
{
  int i = 0;

  if( jinfo_arr != NULL )
  {
    /* We are not starting at the front of the array, so we need to find the 
     * element to start with.
     */
    if( start != NULL )
    {
      for( ; jinfo_arr[i] != NULL && jinfo_arr[i] != start; i++ )
	;
    }
    else
      i = 0;
      
    if( jinfo_arr[i] != NULL )
    {
      if( start_where == START_BEFORE_JOB )
	i--;
      else if( start_where == START_AFTER_JOB )
	i++;

      for(  ; jinfo_arr[i] != NULL; i++ )
      {
	jinfo_arr[i] -> can_not_run = 1;
	update_job_comment(pbs_sd, jinfo_arr[i], comment);
      }
    }
  }
}

/*
 *
 *	job_filter - filters jobs on specified argument
 *
 *	  jobs - array of jobs to filter through
 *	  size - amount of jobs in array
 *	  filter_func - pointer to a function that will filter
 *		- returns 1: job will be added to new array
 *		- returns 0: job will not be added to new array
 *	  arg - an extra arg to pass to filter_func
 *
 *	returns pointer to filtered list
 *
 *	NOTE: this function allocates a new array
 *
 *	filter_func prototype: int func( job_info *, void * )
 *
 */
job_info **job_filter( job_info** jobs, int size,
			int (*filter_func) (job_info*, void*), void *arg)
{
  job_info **new_jobs = NULL;			/* new array of jobs */
  int i, j = 0;

  if( ( new_jobs = malloc( (size + 1) * sizeof( job_info * ) ) ) == NULL )
  {
    perror( "Memory allocation error" );
    return NULL;
  }

  for( i = 0; i < size; i++)
  {
    if( filter_func( jobs[i], arg ) )
    {
      new_jobs[j] = jobs[i];
      j++;
    }
  }
  if( (new_jobs = realloc(new_jobs, (j+1) * sizeof( job_info * ) ) ) == NULL )
  {
    perror( "Memory Allocation Error" );
    return NULL;
  }

  new_jobs[j] = NULL;

  return new_jobs;
}

/*
 *
 *	translate_job_fail_code - translate the failure code of 
 *				is_ok_to_run_job into a comment and log message
 *
 *	  fail_code        - return code from is_ok_to_run_job()
 *	  OUT: comment_msg - translated comment
 *	  OUT: log_msg     - translated log message	
 *
 *	returns 
 *	   1: comment and log messages were set
 *	   0: comment and log messages were not set
 *
 */
int translate_job_fail_code( int fail_code, char *comment_msg, char *log_msg )
{
  int rc = 1;

  if( fail_code < num_res )
  {
    strcpy(comment_msg, res_to_check[fail_code].comment_msg);
    strcpy(log_msg, res_to_check[fail_code].debug_msg);
  }
  else
  {
    switch( fail_code )
    {
      case QUEUE_JOB_LIMIT_REACHED:
        strcpy(comment_msg, COMMENT_QUEUE_JOB_LIMIT);
	strcpy(log_msg, INFO_QUEUE_JOB_LIMIT);
      break;

      case SERVER_JOB_LIMIT_REACHED:
        strcpy(comment_msg, COMMENT_SERVER_JOB_LIMIT);
	strcpy(log_msg, INFO_SERVER_JOB_LIMIT);
      break;

      case SERVER_USER_LIMIT_REACHED:
        strcpy(comment_msg, COMMENT_SERVER_USER_LIMIT);
	strcpy(log_msg, INFO_SERVER_USER_LIMIT);
      break;

      case QUEUE_USER_LIMIT_REACHED:
        strcpy(comment_msg, COMMENT_QUEUE_USER_LIMIT);
	strcpy(log_msg, INFO_QUEUE_USER_LIMIT);
      break;

      case QUEUE_GROUP_LIMIT_REACHED:
        strcpy(comment_msg, COMMENT_QUEUE_GROUP_LIMIT);
	strcpy(log_msg, INFO_QUEUE_GROUP_LIMIT);
      break;

      case SERVER_GROUP_LIMIT_REACHED:
        strcpy(comment_msg, COMMENT_SERVER_GROUP_LIMIT);
	strcpy(log_msg, INFO_SERVER_GROUP_LIMIT);
      break;

      case CROSS_DED_TIME_BOUNDRY:
        strcpy(comment_msg, COMMENT_CROSS_DED_TIME);
	strcpy(log_msg, INFO_CROSS_DED_TIME);
      break;

      case NO_AVAILABLE_NODE:
        strcpy(comment_msg, COMMENT_NO_AVAILABLE_NODE);
	strcpy(log_msg, INFO_NO_AVAILABLE_NODE);
      break;

      case NOT_QUEUED:
	/* do we really care if the job is not in a queued state?  there is 
	 * no way it'll be run in this state, why spam the log about it
	 */
	log_msg[0] = '\0';
	comment_msg[0] = '\0';
	rc = 0;
      break;
      case NOT_ENOUGH_NODES_AVAIL:
        strcpy(comment_msg, COMMENT_NOT_ENOUGH_NODES_AVAIL);
	strcpy(log_msg, INFO_NOT_ENOUGH_NODES_AVAIL);
      break;
      case JOB_STARVING:
	strcpy(comment_msg, COMMENT_JOB_STARVING);
	sprintf(log_msg, INFO_JOB_STARVING, cstat.starving_job -> name);
      break;
      case SCHD_ERROR:
        strcpy(comment_msg, COMMENT_SCHD_ERROR);
	strcpy(log_msg, INFO_SCHD_ERROR);
      break;
      default:
        rc = 0;
        comment_msg[0] = '\0';
	log_msg[0] = '\0';
    }
  }
  return rc;
}

/* 
 *
 *	calc_assn_resource - calcualte the assigned resource in a job array
 *				
 *	  jinfo_arr - array of jobs 
 *	  resstr    - resource to calculate
 *
 *	returns the calculated resource
 *		-1 on error
 *
 */
int calc_assn_resource(job_info **jinfo_arr, char *resstr)
{
  resource_req *req;
  int res_amm = 0;
  int i;

  if( resstr == NULL || jinfo_arr == NULL )
    return -1;

  for(i = 0; jinfo_arr[i] != NULL; i++)
  {
    req = find_resource_req(jinfo_arr[i] -> resreq, resstr);
    if(req != NULL)
      res_amm += req -> amount;
  }

  return res_amm;
}