schedule.c

openPBS的开放源代码

    schd_SubmitQueue->queue->jobs = NULL;

    /* 
     * Any jobs queued on the special queue are now moved to the tail of the
     * newly empty SubmitQueue list.  Mark the jobs as high priority and
     * waiting.
     */
    for (job = queue->jobs; job != NULL; job = nextjob) {

	/*
	 * Keep track of the next job -- the next pointer on this job will
	 * be modified by the schd_move_job_to() function.  Also keep a
	 * pointer to the last job in the list.
	 */
	nextjob = job->next;
	specialtail = job;

	if (job->state != 'Q')
	    continue;

	job->flags |= (JFLAGS_WAITING | JFLAGS_PRIORITY);

	/* 
	 * Move the job from the special queue to the tail of the submit
	 * queue.  This keeps the counts of the queued jobs correct in both
	 * queues.
	 */
	schd_move_job_to(job, schd_SubmitQueue->queue);

	fixedup ++;
    }

    /*
     * Now that the submit queue has the list of all queued jobs from the 
     * special queue, attach the original submit jobs onto the tail of the
     * special job list.
     */
    specialtail->next = submitjobs;

    DBPRT(("%s: fixed up %d jobs.\n", id, fixedup));

    return (fixedup);
}

static int
schedule_restart(Job *joblist)
{
    char   *id = "schedule_restart";
    Job    *job, *nextjob;
    QueueList *qptr;
    int     found, changed;

    changed = found = 0;

    for (job = joblist; job != NULL; job = nextjob) {
	nextjob = job->next;

	if (job->state != 'Q')
	    continue;

	/* 
	 * See if the job is queued on one of the batch queues.  If not,
	 * go on to the next job.
	 */
	for (qptr = schd_BatchQueues; qptr != NULL; qptr = qptr->next)
	    if (strcmp(qptr->queue->qname, job->qname) == 0)
		break;

	if (qptr == NULL)
	    continue;

	found++;

	if (schd_SCHED_RESTART_ACTION == SCHD_RESTART_RERUN) {
	    (void)sprintf(log_buffer, "Restart job '%s' on queue '%s'.",
		job->jobid, job->qname);
	    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, 
		id, log_buffer);
	    DBPRT(("%s: %s\n", id, log_buffer));

	    schd_comment_job(job, schd_JobMsg[JOB_RESTARTED],
		JOB_COMMENT_REQUIRED);

	    if (schd_run_job_on(job, job->queue, schd_SCHED_HOST, 
		LEAVE_JOB_COMMENT)) 
	    {
		(void)sprintf(log_buffer,
		    "Unable to run job '%s' on queue '%s'.", job->jobid,
			job->qname);
		log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id,
		    log_buffer);
	    } else
		changed ++;

	} else /* (SCHED_RESTART_ACTION == SCHD_RESTART_RESUBMIT) */ {
	    if (schd_TEST_ONLY) {
		DBPRT(("%s: would have moved %s back to queue %s\n", id,
		    job->jobid, schd_SubmitQueue->queue->qname));
	    } else {
		/* Move the job back to its originating queue. */
		if (pbs_movejob(connector, job->jobid,
		    job->oqueue, NULL) != 0)
		{
		    (void)sprintf(log_buffer, 
			"failed to move %s to queue %s, %d", job->jobid, 
			job->oqueue, pbs_errno);
		    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, 
			log_buffer);
		    DBPRT(("%s: %s\n", id, log_buffer));
		} else {
		    (void)sprintf(log_buffer, 
			"Requeued job '%s' on queue '%s'.", job->jobid, 
			job->oqueue);
		    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, 
			id, log_buffer);
		    DBPRT(("%s: %s\n", id, log_buffer));
		    schd_comment_job(job, schd_JobMsg[JOB_RESUBMITTED],
			JOB_COMMENT_REQUIRED);
		    changed ++;
		}
	    }
	}
    }

    if (found) {
	if (schd_SCHED_RESTART_ACTION == SCHD_RESTART_RERUN) {
	    (void)sprintf(log_buffer, 
		"Re-ran %d jobs (of %d) found queued on run queues.\n", 
		    changed, found);
	} else {
	    (void)sprintf(log_buffer, 
		"Moved %d queued jobs (of %d) from run queues back to '%s'.\n", 
		    changed, found, schd_SubmitQueue->queue->qname);
	}
	log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);
	DBPRT(("%s: %s\n", id, log_buffer));
    }

    return (changed);
}

static Job *
reject_unrunnables(Job *jobs)
{
    Job *this, *nextjob;
    char tmpstr[300];

    for (this = jobs; this != NULL; this = nextjob) {
	nextjob = this->next;

	if (!schd_job_can_queue(this)) {

	    /*
	     * If this job is at the head of the list, we must deal with
	     * it specially.  We need to advance the list pointer forward
	     * so that further scheduling will not be done on the now
	     * bogus job.  Advance 'jobs', and make 'nextjob' the 'next'
	     * pointer for the new head of the list.
	     */
	    if (this == jobs) {
		jobs = jobs->next;
		nextjob = jobs ? jobs->next : NULL;
	    }

	    DBPRT(("job %s does not fit on any execution queue - reject\n",
		this->jobid));
	    schd_reject_job(this, 
		"Job will not fit on any execution queue.\n"
		"\n"
		"Use 'qstat -q' to get execution queue limits.\n");

	    continue;
	}

	/*
	 * Enforce maximum job limits 
	 * "Big" jobs are given a maximum walltime limit (WALLT_LARGE_LIMIT)
	 * that differs from "small" jobs. (Job size distinction based on
	 * the size specified by SMALL_JOB_MAX.) We need to reject any job
	 * which violate these limits.
	 *
	 * Special-priority jobs are not affected.
	 */
	if (!(this->flags & JFLAGS_PRIORITY) && (schd_SMALL_JOB_MAX > 0)) {
	    if (this->nodes <= schd_SMALL_JOB_MAX) {
		if (this->walltime > schd_WALLT_SMALL_LIMIT) {
		    if (this == jobs) {
			jobs = jobs->next;
			nextjob = jobs ? jobs->next : NULL;
		    }
		    DBPRT(("job %s exceeds Small job walltime limit - reject\n",
			this->jobid));
		    sprintf(tmpstr,
			"Job exceeds maximum walltime limit (%s) policy\n"
			"\tfor small jobs (1 - %d nodes).\n",
			schd_sec2val(schd_WALLT_SMALL_LIMIT),
			schd_SMALL_JOB_MAX);
		    schd_reject_job(this, tmpstr);
		    continue;
		}
	    } else {
		if (this->walltime > schd_WALLT_LARGE_LIMIT) {
		    if (this == jobs) {
			jobs = jobs->next;
			nextjob = jobs ? jobs->next : NULL;
		    }
		    DBPRT(("job %s exceeds Large job walltime limit - reject\n",
			this->jobid));
		    sprintf(tmpstr,
			"Job exceeds maximum walltime limit (%s) policy\n"
			"\tfor large jobs (%d+ nodes).\n",
			schd_sec2val(schd_WALLT_LARGE_LIMIT),
			schd_SMALL_JOB_MAX+1);
		    schd_reject_job(this, tmpstr);

		    continue;
		}
	    }
	}
    }

    return (jobs);
}

#ifdef NODEMASK
static int 
nodemask_overlaps(void)
{
    char    *id = "nodemask_overlaps";
    QueueList *qlp1, *qlp2;
    Bitfield overlap;

    /* Check for overlapping queue nodemasks in batch queues. */
    for (qlp1 = schd_BatchQueues; qlp1 != NULL; qlp1 = qlp1->next) {

	if (!(qlp1->queue->flags & QFLAGS_NODEMASK))
	    continue;

	for (qlp2 = qlp1->next; qlp2 != NULL; qlp2 = qlp2->next) {

	    /* 
	     * If the queues are on different hosts, they may appear to
	     * overlap - no problem.
	     */
	    if (strcmp(qlp1->queue->exechost, qlp2->queue->exechost))
		continue;

	    if ((qlp2->queue->flags & QFLAGS_NODEMASK)) {
		DBPRT(("comparing nodemask for queue %s with queue %s\n",
		    qlp1->queue->qname, qlp2->queue->qname));

		BITFIELD_CPY(&overlap, &(qlp1->queue->queuemask));
		BITFIELD_ANDM(&overlap, &(qlp2->queue->queuemask));
		if (!BITFIELD_IS_ZERO(&overlap)) {
		    (void)sprintf(log_buffer, 
			"Queues %s and %s have overlapping nodemasks!",
			qlp1->queue->qname, qlp2->queue->qname);
		    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, 
			log_buffer);
		    DBPRT(("%s: %s\n", id, log_buffer));
		    return (1);
		}
	    }
	}
    }

    /* Make sure batch queues and external queues have differing nodemasks. */
    for (qlp1 = schd_BatchQueues; qlp1 != NULL; qlp1 = qlp1->next) {

	if (!(qlp1->queue->flags & QFLAGS_NODEMASK))
	    continue;

	for (qlp2 = schd_ExternQueues; qlp2 != NULL; qlp2 = qlp2->next) {

	    /* 
	     * If the queues are on different hosts, they may appear to
	     * overlap - no problem.
	     */
	    if (strcmp(qlp1->queue->exechost, qlp2->queue->exechost))
		continue;

	    if ((qlp2->queue->flags & QFLAGS_NODEMASK)) {
		DBPRT(("comparing nodemask for queue %s with queue %s\n",
		    qlp1->queue->qname, qlp2->queue->qname));

		BITFIELD_CPY(&overlap, &(qlp1->queue->queuemask));
		BITFIELD_ANDM(&overlap, &(qlp2->queue->queuemask));
		if (!BITFIELD_IS_ZERO(&overlap)) {
		    (void)sprintf(log_buffer, 
			"Queues %s and %s have overlapping nodemasks!",
			qlp1->queue->qname, qlp2->queue->qname);
		    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, 
			log_buffer);
		    DBPRT(("%s: %s\n", id, log_buffer));
		    return (1);
		}
	    }
	}
    }

    return (0);
}
#endif	/* NODEMASK */

static int 
make_job_dump (char *dumpfile)
{
    char    *id = "make_job_dump";
    FILE    *dump;
    QueueList *qptr;

    /* 
     * Attempt to open the dump file, creating it if necessary.  It should
     * be truncated each time this runs, so don't open with append mode.
     */

    if ((dump = fopen(dumpfile, "w")) == NULL) {
	(void)sprintf(log_buffer, "Cannot write to %s: %s\n", dumpfile,
	    strerror(errno));
	log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);
	DBPRT(("%s: %s\n", id, log_buffer));
	return (-1);
    }

    /* Head the file with a timestamp. */
    fprintf(dump, "%s\n", ctime(&schd_TimeNow));

    /* Include the version string compiled into the scheduler binary. */
    fprintf(dump, "%s\n", schd_VersionString);

    /* And some more useful information about the state of the world. */
    fprintf(dump, "Scheduler running on '%s'\n", schd_ThisHost);

    fprintf(dump, "Prime-time is ");
    if (schd_ENFORCE_PRIME_TIME && schd_TimeNow >= schd_ENFORCE_PRIME_TIME) {
	fprintf(dump, "from %s ", schd_sec2val(schd_PRIME_TIME_START));
	fprintf(dump, "to %s.\n", schd_sec2val(schd_PRIME_TIME_END));
    } else
	fprintf(dump, "not enforced.\n");

    fprintf(dump, "\nJOBS LISTED IN ORDER FROM HIGHEST TO LOWEST PRIORITY\n\n");

    /* Now dump the jobs queued on the various queues, in order of priority. */
    qptr = schd_SubmitQueue;
    if (qptr->queue->jobs) {
	fprintf(dump, "Jobs on submit queue '%s':\n", qptr->queue->qname);
	dump_sorted_jobs (dump, qptr->queue->jobs);
    }

    for (qptr = schd_ExternQueues; qptr != NULL; qptr = qptr->next) {
	if (qptr->queue->jobs) {
	    fprintf(dump, "Jobs on external queue '%s':\n", qptr->queue->qname);
	    dump_sorted_jobs (dump, qptr->queue->jobs);
	}
    }
    for (qptr = schd_DedQueues; qptr != NULL; qptr = qptr->next) {
	if (qptr->queue->jobs) {
	    fprintf(dump, "Jobs on dedicated queue '%s':\n", 
		qptr->queue->qname);
	    dump_sorted_jobs (dump, qptr->queue->jobs);
	}
    }

    if (fclose(dump)) {
	(void)sprintf(log_buffer, "close(%s): %s\n", dumpfile, strerror(errno));
	log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);
	DBPRT(("%s: %s\n", id, log_buffer));
	return (-1);
    }
    return (0);
}

static int 
dump_sorted_jobs (FILE *dump, Job *joblist)
{
    Job     *job;
    int      njobs;
    int      elig_mesg = 0;

#define DUMP_JID_LEN		16
#define DUMP_STATE_LEN		1
#define DUMP_OWNER_LEN		8
#define DUMP_NODES_LEN		3
#define DUMP_WALLT_LEN		8
#define DUMP_WAITT_LEN		8
#define DUMP_ELIGI_LEN		9	/* time plus '*' if wait != eligible */
#define DUMP_FLAGS_LEN		18

    char     jid[DUMP_JID_LEN + 1];
    char     owner[DUMP_OWNER_LEN + 1];
    char     wallt[DUMP_WALLT_LEN + 1];
    char     waitt[DUMP_WAITT_LEN + 1];
    char     eligi[DUMP_ELIGI_LEN + 1];
    char     flags[DUMP_FLAGS_LEN + 1];

    fprintf(dump, "  %*s %*s %*s %*s %*s %*s %*s %*s\n",
	-DUMP_JID_LEN,		"Job ID", 
	-DUMP_STATE_LEN,	"S", 
	-DUMP_OWNER_LEN,	"Owner", 
	-DUMP_NODES_LEN,	"Nds", 
	-DUMP_WALLT_LEN,	"Walltime", 
	-DUMP_WAITT_LEN,	"Q'd for", 
	-DUMP_ELIGI_LEN,	"Eligible", 
	-DUMP_FLAGS_LEN,	"Flags");

    fprintf(dump, "  %*s %c %*s %*s %*s %*s %*s %*s\n",
	-DUMP_JID_LEN,		"----------------", 
	'-', 
	-DUMP_OWNER_LEN,	"--------", 
	-DUMP_NODES_LEN,	"---", 
	-DUMP_WALLT_LEN,	"--------", 
	-DUMP_WAITT_LEN,	"--------", 
	-DUMP_ELIGI_LEN,	"---------", 
	-DUMP_FLAGS_LEN,	"------------------");

    for (njobs = 0, job = joblist; job != NULL; job = job->next) {

	njobs++;
	strncpy(jid, job->jobid, DUMP_JID_LEN);
	strncpy(owner, job->owner, DUMP_OWNER_LEN);
	strcpy(wallt, schd_sec2val(job->walltime));
	strcpy(waitt, schd_sec2val(job->time_queued));
	strcpy(eligi, schd_sec2val(job->eligible));
	if (job->time_queued != job->eligible) {
	    strcat(eligi, "*");
	    elig_mesg ++;
	}

	flags[0] = '\0';

	/* Watch length of 'flags[]' array! */
	if (job->flags & JFLAGS_INTERACTIVE)
	    strcat(flags, "Int ");

	/* "Priority" jobs are marked as being waiting, even if they're new. */
	if (job->flags & JFLAGS_PRIORITY)
	    strcat(flags, "High ");
	else if (job->flags & JFLAGS_WAITING)
	    strcat(flags, "Wait ");

	if (job->flags & JFLAGS_DEDICATED)
	    strcat(flags, "Ded ");

	if (job->flags & JFLAGS_NEEDS_HPM)
	    strcat(flags, "HPM ");

	/* Trim off the trailing space if any flags were listed. */
	if (flags[0] != '\0')
	    flags[strlen(flags) - 1] = '\0';

	fprintf(dump, "  %*s %c %*s %*d %*s %*s %*s %*s\n",
	    -DUMP_JID_LEN,	jid,
	    job->state,
	    -DUMP_OWNER_LEN,	job->owner,
	    -DUMP_NODES_LEN,	job->nodes,
	    -DUMP_WALLT_LEN,	wallt,
	    -DUMP_WAITT_LEN,	waitt,
	    -DUMP_ELIGI_LEN,	eligi,
	    -DUMP_FLAGS_LEN,	flags);
    }

    fprintf(dump, "    Total: %d job%s\n\n", njobs, (njobs == 1) ? "" : "s");

    if (elig_mesg) {
	fprintf(dump, "Jobs marked with a ``*'' have an etime different "
	    "from their ctime.\n\n");
    }

    return (njobs);
}