usersort.c

openPBS的开放源代码

    /*
     * "Seed" the linked list by pointing to a bogus initial element.
     * Since the jobtail->next pointer will always be valid (either it
     * hangs off the seed or a real job) this simplifies the following
     * list operations considerably.
     */
    joblist = &list_seed;
    jobtail = &list_seed;
    jobtail->next = NULL;

    /* Walk the running jobs and place them on the list. */
    for (i = 0; i < nJRs; i++)
	jobtail = jobtail->next = runningJobs[i];

    /* Walk the normal jobs and place them on the list. */
    for (i = 0; i < nnormalQ; i++)
	jobtail = jobtail->next = normalQ[i];

    /* Place any remaining jobs on the end of the list. */
    for (i = 0; i < notherQ; i++)
	jobtail = jobtail->next = otherQ[i];

    /* Terminate the last element on the list with a NULL next pointer. */
    jobtail->next = NULL;

    /* Free any storage allocated for the lists. */
    if (runningJobs)
	free(runningJobs); 

    if (normalQ)
	free(normalQ);

    if (otherQ)
	free(otherQ);

    /* And reset all the values. */
    runningJobs = normalQ = otherQ = NULL;
    nJRs = nJQs = nnormalQ = notherQ = 0;

    /*
     * The first element on joblist is the pointer to the list_seed.  It's
     * next pointer points to the head of the real list - return that.
     */
    return (joblist->next);
}

void create_job_schedule()
{
    char   *id = "create_job_schedule";
    Job    *job_ptr;
    int     i;

    /*
     * Destroy any previously created list.
     */
    if (JobSchedule)
	free(JobSchedule);

    JobSchedule  = NULL;
    nJobSchedule = 0;

    /* 
     * Walk the list of "running" jobs, adding Start and End entries
     * to the global JobSchedule table along the way.
     */
    for (i = 0; i < nJRs; i++) {

	/* Start Entry */
	++nJobSchedule;
	JobSchedule = realloc(JobSchedule, nJobSchedule * sizeof *JobSchedule);
	if (!JobSchedule) {
	    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER,
	        id, "realloc(JobSchedule)");
	    return;
	}
	JobSchedule[nJobSchedule -1].event    = 'S';
	JobSchedule[nJobSchedule -1].time     = runningJobs[i]->mtime;
	JobSchedule[nJobSchedule -1].cpu      = runningJobs[i]->ncpus;
	JobSchedule[nJobSchedule -1].mem      = runningJobs[i]->memory;
	JobSchedule[nJobSchedule -1].walltime = runningJobs[i]->walltime +2*60;
	strcpy(JobSchedule[nJobSchedule -1].qname, runningJobs[i]->qname);
	strcpy(JobSchedule[nJobSchedule -1].group, runningJobs[i]->group);
	strcpy(JobSchedule[nJobSchedule -1].arch, 
	    get_queue_arch(runningJobs[i]->qname));

	/* End Entry */
	++nJobSchedule;
	JobSchedule = realloc(JobSchedule, nJobSchedule * sizeof *JobSchedule);
	if (!JobSchedule) {
	    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER,
	        id, "realloc(JobSchedule)");
	    return;
	}
	JobSchedule[nJobSchedule -1].event 	= 'E';
	JobSchedule[nJobSchedule -1].time 	= 
		schd_TimeNow + runningJobs[i]->time_left + 2*60;
	JobSchedule[nJobSchedule -1].cpu 	= runningJobs[i]->ncpus;
	JobSchedule[nJobSchedule -1].mem 	= runningJobs[i]->memory;
	JobSchedule[nJobSchedule -1].walltime 	= 0;
	strcpy(JobSchedule[nJobSchedule -1].qname, runningJobs[i]->qname);
	strcpy(JobSchedule[nJobSchedule -1].group, runningJobs[i]->group);
	strcpy(JobSchedule[nJobSchedule -1].arch, 
	    get_queue_arch(runningJobs[i]->qname));
    }

    /* now sort the table into chronological order */
    qsort(JobSchedule, nJobSchedule, sizeof *JobSchedule, compare_qtime);

    schd_print_schedule();

    return;
}

void schd_print_schedule()
{
    int i;

    /* print Schedule for sanity check */
#if 0
    printf("DEBUG: (job schedule)\n");
    for (i=0; i<nJobSchedule; i++) {
	printf("%c %ld %s %d cpus %ld(b)memory %ld\n", JobSchedule[i].event,
		JobSchedule[i].time, JobSchedule[i].qname, JobSchedule[i].cpu,
		schd_byte2val(JobSchedule[i].mem), JobSchedule[i].walltime);
    }
#endif
}

void schd_dump_schedule(QueueList *qlist)
{
    int i, avail_cpu;
    size_t avail_mem;
    Queue  *queue, *best_queue;
    QueueList *qptr;
    FILE *fptr;

    fptr = fopen(schd_PREDICTED_SCHEDULE, "w");
    if (fptr == NULL) {
	;
    }

    for (qptr = qlist; qptr != NULL; qptr = qptr->next) {
        queue = qptr->queue;

	if (queue->rsrcs == NULL)
	    continue;

	avail_cpu = queue->rsrcs->ncpus_total;
	avail_mem = queue->rsrcs->mem_total;
        for (i=0; i<nJobSchedule; i++) {
	    if (!strcmp(queue->qname,JobSchedule[i].qname)) {
		if (JobSchedule[i].event == 'S') {
		    avail_cpu -= JobSchedule[i].cpu;
		    avail_mem -= JobSchedule[i].mem;
		}
		if (JobSchedule[i].event == 'E') {
		    avail_cpu += JobSchedule[i].cpu;
		    avail_mem += JobSchedule[i].mem;
		}
	        fprintf(fptr, "%s %s %ld %d cpus %ld mem\n",
		    queue->rsrcs->arch, queue->qname, JobSchedule[i].time,
		    avail_cpu, avail_mem);
	    }
	}
    }
    fclose(fptr);
}

void schd_update_schedule(Job *job, Queue *queue, time_t start, char *reason)
{
    char   *id = "update_schedule";
    char   tmpstring[500];
    char   tmpstring2[500];
    char  *pstr;
    struct tm *tmptr;
    int i;

    /* Start Entry */
    ++nJobSchedule;
    JobSchedule = realloc(JobSchedule, nJobSchedule * sizeof *JobSchedule);
    if (!JobSchedule) {
        log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER,
            id, "realloc(JobSchedule)");
        return;
    }
    JobSchedule[nJobSchedule -1].event       = 'S';
    JobSchedule[nJobSchedule -1].time        = start;
    JobSchedule[nJobSchedule -1].cpu         = job->ncpus;
    JobSchedule[nJobSchedule -1].mem         = job->memory;
    JobSchedule[nJobSchedule -1].walltime    = job->walltime + 2*60;
    strcpy(JobSchedule[nJobSchedule -1].qname, queue->qname);
    strcpy(JobSchedule[nJobSchedule -1].group, job->group);
    strcpy(JobSchedule[nJobSchedule -1].arch, 
	    get_queue_arch(queue->qname));

    /* End Entry */
    ++nJobSchedule;
    JobSchedule = realloc(JobSchedule, nJobSchedule * sizeof *JobSchedule);
    if (!JobSchedule) {
        log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER,
            id, "realloc(JobSchedule)");
        return;
    }
    JobSchedule[nJobSchedule -1].event       = 'E';
    JobSchedule[nJobSchedule -1].time        = start + job->walltime;
    JobSchedule[nJobSchedule -1].cpu         = job->ncpus;
    JobSchedule[nJobSchedule -1].mem         = job->memory;
    JobSchedule[nJobSchedule -1].walltime    = 0;
    strcpy(JobSchedule[nJobSchedule -1].qname, queue->qname);
    strcpy(JobSchedule[nJobSchedule -1].group, job->group);
    strcpy(JobSchedule[nJobSchedule -1].arch, 
	    get_queue_arch(queue->qname));

    /* now sort the table into chronological order */
    qsort(JobSchedule, nJobSchedule, sizeof *JobSchedule, compare_qtime);

    /* finially, update the job comment */

    if (reason) {
        if ((pstr = strstr(reason, "Est.")) != NULL)
            strcpy(tmpstring2, reason+16);
        else
            strcpy(tmpstring2, reason);
    } else
        sprintf(tmpstring2, "%s (%s)", schd_JobMsg[NO_ARCH], job->arch);
  
    tmptr = localtime(&start);
    sprintf(tmpstring, "Est.%0.2d/%0.2d %0.2d:%0.2d] %s",
        tmptr->tm_mon+1, tmptr->tm_mday, tmptr->tm_hour,
        tmptr->tm_min, tmpstring2);
    strcpy(reason, tmpstring);
}

time_t schd_when_can_job_start(Job *job, Queue *queue, char *reason)
{
    char *id = "when_can_job_start";
    int i, count, now_cpu, now_run, can_run, arch_cpu, save_arch_cpu;
    size_t now_mem, arch_mem, save_arch_mem;
    time_t steptime, lasttime, possible_start;

    count   = 0;
    now_cpu = 0;
    now_mem = 0;
    now_run = 0;
    arch_cpu = 0;
    arch_mem = 0;
    save_arch_cpu = 0;
    save_arch_mem = 0;
    lasttime = 0;
    steptime = 0;
    possible_start = 0;

    /*
     * walk schedule looking for a window in time in which we could
     * run this job on this queue/exechost
     */

    for (i=0; i<nJobSchedule; i++) {

	/* Count up running jobs for the requested arch; we'll need
	 * this info later if we find a matching slot/host.
	 */

        if ((!strcmp(JobSchedule[i].arch, queue->rsrcs->arch)) &&
           (!strcmp(JobSchedule[i].group, job->group))) {
            if (JobSchedule[i].event == 'S') {
	        arch_cpu += JobSchedule[i].cpu;
	        arch_mem += JobSchedule[i].mem;
	    }

	    if (JobSchedule[i].event == 'E') {
	        arch_cpu -= JobSchedule[i].cpu;
		if (arch_cpu < 0)
		    arch_cpu = 0;

		if (arch_mem < JobSchedule[i].mem)
		    arch_mem = 0;
		else
	            arch_mem -= JobSchedule[i].mem;
	    }

	    if (JobSchedule[i].time <= schd_TimeNow) {
		save_arch_cpu = arch_cpu;
		save_arch_mem = arch_mem;
	    }
	}

        /* Only look at scheduled events for the requested queue */
        if (!strcmp(JobSchedule[i].qname, queue->qname)) {
	    count++;

            if (JobSchedule[i].event == 'S') {
	        now_cpu += JobSchedule[i].cpu;
	        now_mem += JobSchedule[i].mem;
                now_run ++;
	    }

	    if (JobSchedule[i].event == 'E') {
	        now_cpu -= JobSchedule[i].cpu;
	        now_mem -= JobSchedule[i].mem;
	        now_run --;
	    }


 	    can_run = schd_job_could_run(now_cpu, now_mem, now_run,
		arch_mem, arch_cpu, job, queue, reason);

	    if (JobSchedule[i].time <= schd_TimeNow) {
	        if (can_run)
	            possible_start = schd_TimeNow;
		else
	            possible_start = 0;
	    }

	    if (JobSchedule[i].time > schd_TimeNow)  { /* future time */

		if (possible_start) {
 	            if (possible_start + job->walltime + 2*60 <
		        JobSchedule[i].time)
  		        break;

		    if (can_run)
                        steptime = possible_start;

	        } else {
 	            if (can_run)
                        steptime = JobSchedule[i].time;
 	        }

    	        if (can_run)
	            possible_start = steptime;
	        else
		    possible_start = 0;
	    }
	}
    }

    if (nJobSchedule == 0 || count == 0) {
	can_run = schd_job_could_run(0, 0, 0, save_arch_mem, save_arch_cpu,
	    job, queue, reason);

        if (can_run)
	    possible_start = schd_TimeNow;
    }

    if (possible_start != 0 && possible_start < schd_TimeNow)
	    possible_start = schd_TimeNow;

    return(possible_start);
}