getqueues.c

openPBS的开放源代码

    /*
     * Find out if the queue is idle, and if it was not before, set the idle
     * time to now.  If there are running jobs, the queue is not idle at the
     * start of this iteration - set idle_since to 0.
     */
    if (queue->running) {
	queue->idle_since = 0;
    } else {
	if (queue->idle_since == 0)
	    queue->idle_since = schd_TimeNow;
    }

    /*
     * Get the resources for this queue from the resource monitor (if
     * available).  If the resmom is not accessible, disable the queue.
     * If the resources were received okay, compute the available node
     * masks from the resources and jobs.
     * Don't bother with resources for the submit queue.
     */
    if (strcmp(queue->qname, schd_SubmitQueue->queue->qname) != 0)
    {
	queue->rsrcs = schd_get_resources(queue->exechost);
	if (queue->rsrcs != NULL) {
	    /* Account for this queue's resources. */
	    queue->rsrcs->nodes_alloc += queue->nodes_assn;
	    queue->rsrcs->njobs       += queue->running;

	} else {
	    (void)sprintf(log_buffer,
		"Can't get resources for %s@%s - marking unavailable.",
		queue->qname, queue->exechost);
	    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);
	    DBPRT(("%s: %s\n", id, log_buffer));

	    queue->flags |= QFLAGS_DISABLED;
	}
    }

#ifdef DEBUG
    schd_dump_queue(queue, QUEUE_DUMP_JOBS);
#endif /* DEBUG */

    /*
     * It would probably be better to wait for the world to stabilize
     * than to try to impose some artificial order upon it.  Do not do
     * the sanity check if the queue is stopped.
     */
    if ((queue->flags & QFLAGS_STOPPED) == 0) {
	if (!queue_sanity(queue)) {
	    sprintf(log_buffer, "WARNING! Queue '%s' failed sanity checks.",
		queue->qname);
	    log_record(PBSEVENT_ERROR, PBS_EVENTCLASS_SERVER, id, log_buffer);
	    DBPRT(("%s: %s\n", id, log_buffer));

	    return (1);
	}
    }

    return (0);
}

#ifdef DEBUG
void 
schd_dump_queue(Queue *queue, int dumpjobs)
{
    Job    *job;
    UserAcl *aclent;
    char    num[32];
    char   *ptr;
    int     columns;

    DBPRT(("\nQueue '%s@%s': %sabled/%sed", 
	queue->qname, queue->exechost,
	(queue->flags & QFLAGS_DISABLED) ? "Dis" : "En",
	(queue->flags & QFLAGS_STOPPED) ? "Stopp" : "Start"));

    DBPRT(("%s%s%s%s ",
	(queue->flags & QFLAGS_FULL) ? "/Full" : "",
	(queue->flags & QFLAGS_MAXRUN) ? "/MaxRun" : "",
	(queue->flags & QFLAGS_DRAINING) ? "/Drain" : "",
	(queue->flags & QFLAGS_USER_ACL) ? "/ACL" : ""));

    if (schd_ENFORCE_PRIME_TIME && schd_TimeNow >= schd_ENFORCE_PRIME_TIME)
	DBPRT(("obsv_pt:%s", queue->observe_pt ? "Yes" : "No"));
    DBPRT(("\n"));

    sprintf(num, "%d", queue->running);
    DBPRT(("  Job counts: %s running, ", 
	queue->running != UNSPECIFIED ? num : "???"));

    sprintf(num, "%d", queue->maxrun);
    DBPRT(("%s max ", queue->maxrun != UNSPECIFIED ? num : "???"));

    sprintf(num, "%d", queue->userrun);
    DBPRT(("(%s/user), ", queue->userrun != UNSPECIFIED ? num : "???"));

    sprintf(num, "%d", queue->queued);
    DBPRT(("%s queued\n", queue->queued != UNSPECIFIED ? num : "???"));

    sprintf(num, "%d", queue->nodes_assn);
    DBPRT(("  Nodes:%s/", queue->nodes_assn != UNSPECIFIED ? num : "???"));

    sprintf(num, "%d", queue->nodes_max);
    DBPRT(("%s", queue->nodes_max != UNSPECIFIED ? num : "???"));

    sprintf(num, "%d", queue->nodes_default);
    DBPRT((" [def %s, ", queue->nodes_default != UNSPECIFIED ? num : "???"));

    sprintf(num, "%d", queue->nodes_min);
    DBPRT(("min %s], ", queue->nodes_min != UNSPECIFIED ? num : "???"));

    DBPRT(("wallt max %s ", (queue->wallt_max != UNSPECIFIED) ?
	    schd_sec2val(queue->wallt_max) : "???"));
    DBPRT(("[def %s ", 
        queue->wallt_default!=UNSPECIFIED ?
	    schd_sec2val(queue->wallt_default) : "???"));
    DBPRT(("min %s]\n", (queue->wallt_min != UNSPECIFIED) ?
	    schd_sec2val(queue->wallt_min) : "???"));

    if (queue->empty_by)
	/* ctime(2) returns a '\n'-terminated string, so no additional '\n' */
	DBPRT(("  Queue will empty by: %s", ctime(&queue->empty_by)));

    if (queue->idle_since)
	/* ctime(2) returns a '\n'-terminated string, so no additional '\n' */
	DBPRT(("  Queue idle since: %s", ctime(&queue->idle_since)));

    if (queue->useracl && (queue->flags & QFLAGS_USER_ACL)) {
	DBPRT(("    User ACL: "));
	columns = 9;		/* Start with 9 columns for 'User ACL: ' */
	for (aclent = queue->useracl; aclent != NULL; aclent = aclent->next) {
	    columns += strlen(aclent->user) + 1;
	    if (columns >= 72) {
		DBPRT(("\n    "));
		columns = 0;
	    }
	    DBPRT(("%s%s", 
		((columns == 0) || (aclent == queue->useracl)) ? "" : "/", 
		aclent->user));
	}
	DBPRT(("\n"));
    }

    if (dumpjobs && queue->jobs) {
	DBPRT(("  Jobs: "));

	columns = 5;		/* Start with 5 columns for 'Jobs: ' */

	for (job = queue->jobs; job != NULL; job = job->next) {

	    /* Just the job numbers -- but be sure to put the '.' back! */
	    if ((ptr = strchr(job->jobid, '.')) != NULL)
		*ptr = '\0';
	    
	    columns += strlen(job->jobid) + 3; /* 3 == job->state + '/' + ' ' */

	    if (columns >= 72) {
		DBPRT(("\n   "));
		columns = 0;
	    }

	    DBPRT((" %s/", job->jobid));
	    DBPRT(("%c",
		(job->flags & JFLAGS_PRIORITY) ? '!' :
		(job->flags & JFLAGS_WAITING) ? 'W' :
		job->state));
		
	    if (ptr != NULL)
		*ptr = '.';
	}
	DBPRT(("\n"));
    }
}
#endif /* DEBUG */

static int 
queue_claim_jobs(Queue *queue, Job **joblist_ptr)
{
    Job    *job, *prev, *next, *qtail, *longest;
    int     moved;
    int     running, queued, held, other;

    /* 
     * The time at which this queue should be empty (i.e. when all jobs
     * currently running on it are completed).  Keep track of the longest
     * running job and compute the empties_at value from that.
     */
    longest	= NULL;

    /*
     * Keep track of some statistics about what jobs have been found in the
     * list.  These aren't really used (yet), but are easy to compute.
     */
    running	= 0;
    queued	= 0;
    held	= 0;
    other	= 0;

    /* 
     * The number of jobs that have been moved from the global list to the
     * per-queue list.
     */
    moved	= 0;

    /*
     * Find the last element of the list of jobs on this queue.  This is
     * probably unnecessary (since this should always be called with 
     * 'queue->jobs' pointing to NULL.  Still, it doesn't hurt to try.
     */
    qtail = NULL;
    if (queue->jobs) {
	for (qtail = queue->jobs; qtail->next != NULL; qtail = qtail->next)
	    /* Do nothing - just walk to next-to-last element of list */
	    ;
    }

    prev = NULL;
    for (job = *joblist_ptr; job != NULL; job = next) {
	next = job->next;

	if (strcmp(job->qname, queue->qname)) {
	    /* check if this is a challenge or background job that needs
	     * to be merged with the submit list of jobs 
 	     */
	    if (!strcmp(queue->qname,
 		 schd_SubmitQueue->queue->qname)) {  /* the submit list? */
		if ((strcmp(job->qname, schd_ChallengeQueue)) &&
		    (strcmp(job->qname, schd_SpecialQueue)) &&
		    (strcmp(job->qname, schd_BackgroundQueue))) {

	    		/* Wrong queue -- ignore this job. */
	    		prev = job;
	    		continue;
		}
	    }
	}

	/*
	 * This job belongs to this queue.  Remove it from the job list and
	 * place it at the tail of the queue's job list.  This is somewhat
	 * complicated since we have to remove it from the joblist first.
	 * If there is no "previous" job element, then the current job is
	 * the head of the list.
	 */
	if (job == *joblist_ptr) {

	    /* 
	     * This is the head of the list -- just point the list head to
	     * the job's next pointer and now the job is "orphaned".
	     */


	    *joblist_ptr = next;
	    prev = *joblist_ptr;

	} else {

	    /*
	     * This job lies in the middle of the list somewhere.  Jump over
	     * it in the previous element, and we're done.  Note that since
	     * we skipped this job, the previous job pointer does not change.
	     */
	    
	    prev->next = job->next;
	}

	/* 
	 * Now 'job' is the only active handle on the job.  Place it at the
	 * tail of the queue's list.  If 'qtail' is NULL, this is the first
	 * job -- place it at the head of the list.  Otherwise, place it after
	 * the element pointed to by the 'qtail'.  Either way, this is the
	 * last element in the list, so point the qtail at it and clear its 
	 * next pointer.
	 */
	if (qtail == NULL)
	    queue->jobs = job;
	else
	    qtail->next = job;

	qtail = job;

	job->next = NULL;

	if (job->flags & JFLAGS_QNAME_LOCAL) {
	    /*
	     * The job has some memory that was allocated when it was created,
	     * that is used to store the name of the queue on which it resides.
	     * Since we know exactly what queue it lives on (the one pointed to
	     * by 'queue', to be exact), we can free the storage and point the 
	     * 'job->qname' at 'queue->qname'.  Also store a reference to the
	     * owner queue in the job itself.
	     */
	    free (job->qname);
	    job->qname = queue->qname;
	    job->queue = queue;

	    /* 
	     * Turn off the flag -- job->qname is now a reference to a queue.
	     */
	    job->flags &= ~JFLAGS_QNAME_LOCAL;
	}

	/* Now, count the job and increment the correct statistic. */
	moved ++;

	switch (job->state) {
	    case 'R':	
		running ++;	

		/*
		 * Adjust the time of any job that has run over its expected
		 * time to the JOB_OVERTIME_SLOP.
		 */
		if (job->time_left < 0) {
		    job->time_left = JOB_OVERTIME_SLOP;
		}

		/* If this job will be running the longest, note that. */
		if ((longest == NULL) || 
		    (job->time_left > longest->time_left)) 
		{
		    longest = job;
		}

		break;

	    case 'Q':	
		queued ++;
		break;

	    case 'H':	
		held ++;
		break;

	    default:	
	    	other ++;
		break;
	}

    }

    /*
     * If any jobs were running, then set the empty_by time to the absolute
     * time (in seconds) when all jobs should be completed.  If none are
     * running, then set empty_by time to 0.
     */
    if (longest)
	queue->empty_by = schd_TimeNow + longest->time_left;
    else
	queue->empty_by = 0;

    return (moved);
}

/*
 * PBS has been known to provide some bizarre information about the state
 * of the queue (i.e. no jobs are running in it, but some of its resources
 * are consumed).  Perform some consistency checks on the queue information
 * before agreeing that it is correct.  Generate a laundry list of everything
 * that appears wrong with it.
 */
static int
queue_sanity(Queue *queue)
{
    char   *id = "queue_sanity";
    Job    *job;
    int     queued, running;
    int     is_sane;

    is_sane = 1;	/* Assume the queue is sane for now. */

    /*
     * Count running and queued jobs and make sure the numbers match up.
     */
    queued = running = 0;

    for (job = queue->jobs; job != NULL; job = job->next) {
	switch (job->state) {
	case 'R':
	    running ++;
	    break;

	case 'Q':
	    queued ++;
	    break;

	default:
	    /* Empty */
	    break;
	}
    }

/* JJPJ -- this no longer makes much sense since we are adding 
 * 	   now moving challenge and background jobs into the queue
    if (queue->running != running) {
	sprintf(log_buffer, 
	    "Queue '%s' expected %d running jobs, but found %d",
	    queue->qname, queue->running, running);
	log_record(PBSEVENT_ERROR, PBS_EVENTCLASS_SERVER, id, log_buffer);
	DBPRT(("%s: %s\n", id, log_buffer));
	is_sane = 0;
    }

    if (queue->queued != queued) {
	sprintf(log_buffer, 
	    "Queue '%s' expected %d queued jobs, but found %d",
	    queue->qname, queue->queued, queued);
	log_record(PBSEVENT_ERROR, PBS_EVENTCLASS_SERVER, id, log_buffer);
	DBPRT(("%s: %s\n", id, log_buffer));
	is_sane = 0;
    }
 *
 */

    if (queue->running && (queue->empty_by < schd_TimeNow)) {
	sprintf(log_buffer, 
	    "Queue '%s' was expected to be empty %d seconds ago",
	    queue->qname, (schd_TimeNow - queue->empty_by));
	log_record(PBSEVENT_ERROR, PBS_EVENTCLASS_SERVER, id, log_buffer);
	DBPRT(("%s: %s\n", id, log_buffer));
	is_sane = 0;
    }

    return (is_sane);
}