jobinfo.c

openPBS的开放源代码

	if (!strcmp(attr->name, ATTR_used)) {   
	    if (!strcmp(attr->resource, "walltime")) {
		job->walltime_used = schd_val2sec(attr->value);

		changed ++;
	    }
	    /* No other interesting cases. */
	    continue;
	}

	/* Session ID for running jobs (used to correlate GRM info */
	if (!strcmp(attr->name, ATTR_session)) {
	    job->session = atoi(attr->value);
	    continue;
	}

	/* Job Priority attribute (inherited from queue) */
	if (!strcmp(attr->name, ATTR_p)) {
	    job->priority = atoi(attr->value);
	    continue;
	}

	/* Creation time attribute. */
	if (!strcmp(attr->name, ATTR_ctime)) {
	    /* How long ago was it put in the queue ? */
	    job->time_queued = schd_TimeNow - atoi(attr->value);

	    continue;
	}

	/* Modified time attribute. */
	if (!strcmp(attr->name, ATTR_mtime)) {
	    /* When was the job last modified? */
	    job->mtime = atoi(attr->value);

	    continue;
	}

#ifdef ATTR_etime
	/* 
	 * When was the job last eligible to run?  When a user-hold is
	 * released, this value is updated to the current time.  This
	 * prevents users from gaining higher priority from holding their
	 * jobs.
	 */
	if (!strcmp(attr->name, ATTR_etime)) {
	    job->eligible = schd_TimeNow - atoi(attr->value);

	    continue;
	}
#endif /* ATTR_etime */
    }

    /* 
     * If this job is in the "Running" state, compute how many seconds 
     * remain until it is completed.
     */
    if (job->state == 'R') {
	job->time_left = job->walltime - job->walltime_used;
    }
 
    /*
     * If this job was enqueued since the last time we ran, set the job
     * flag to indicate that we have not yet seen this job.  This makes it
     * a candidate for additional processing.  There may be some inaccuracy,
     * since the time_t has resolution of 1 second.  Attempt to err on the
     * side of caution.
     */
    if ((job->state == 'Q') && (job->time_queued != UNSPECIFIED)) {
        if (job->time_queued <= (schd_TimeNow - schd_TimeLast)) {
	    job->flags |= JFLAGS_FIRST_SEEN;
	}
    }

    /*
     * If the 'etime' attribute wasn't found, set it to the time the job has
     * been queued.  Most jobs will be eligible to run their entire lifetime.
     * The exception is a job that has been held - if it was a user hold,
     * the release will reset the etime to the latest value.
     * If not eligible time was given, use the job's creation time.
     */
    if (!job->eligible)
	job->eligible = job->time_queued;

#if defined(sgi)
    /*
     * If the job provided a memory or CPU resource that does not match
     * the resources that will be allocated by the assigned nodes (i.e.
     * a request for 100mb of memory and 16 CPUs - the job will "get" all
     * 4GB of memory anyway), alter the job attributes such that they
     * will align with the assigned nodes later.
     */
    bump_rsrc_requests(job, cpu_req, mem_req);

#endif /* defined(sgi) */

    /* 
     * Need to update the  time_until_eligible  and  total_delay fields,
     * probably from a global array of information saved from previous
     * scheduler iteration.
     */

    /* 
     * Calculate the job priority weight sort key to be used later in
     * job sorting. This is the "priority" the job should have during
     * sorting based on the size of the job, the length of time queued,
     * and the job type.
     */
    calc_job_weight(job);

    return (changed);
}

/*
 * Walk a list of jobs, freeing any information on them and then freeing
 * each element.  Returns number of elements freed.
 */
int
schd_free_jobs(Job *list)
{
    Job    *job, *next;
    int     numjobs = 0;

    for (job = list; job != NULL; job = next) {
        numjobs ++;
	next = job->next;

	if (job->jobid)		free(job->jobid);
	if (job->owner)		free(job->owner);
	if (job->host)		free(job->host);
	if (job->exechost)	free(job->exechost);
	if (job->group)		free(job->group);
	if (job->comment)	free(job->comment);
	if (job->oqueue)	free(job->oqueue);

	/*
	 * Do *not* free the Job's backpointer to the queue's name if it
	 * is a reference to the Queue's qname.  If it is local storage
	 * created by schd_strdup(), it should be freed to prevent a 
	 * memory leak.
	 */
	if (job->flags & JFLAGS_QNAME_LOCAL)
	    free(job->qname);

	free(job);
    }

    return (numjobs);
}

static int 
bump_rsrc_requests(Job *job, int cpu_req, size_t mem_req)
{
    char   *id = "bump_rsrc_requests";
    char   *val;
    char    buf[64];
    int     bumped = 0;

    /* 
     * If a job gives the "wrong" value for the memory request (for the
     * number of nodes required to fulfill the request), then bump the
     * memory request to the amount of memory the assigned nodes would
     * consume.
     */
    if ((mem_req == 0) || (mem_req != (job->nodes * MB_PER_NODE))) {

	/* Make a printable version of the requested memory. */
	strcpy(buf, schd_byte2val(mem_req));

	/* Compute the "right" memory request, based on the nodes. */
	val = schd_byte2val(job->nodes * MB_PER_NODE);
	if (val == NULL)
	    return (1);

	if (schd_alterjob(connector, job, ATTR_l, val, "mem")) {
	    DBPRT(("%s: Failed to set job %s \"mem\" attr to %s\n", id,
		job->jobid, val));
	    return (1);
	}

	bumped++;
    }

    /* 
     * If a job gives the "wrong" value for the CPU request (for the
     * number of nodes required to fulfill the request), then bump the
     * CPU request to the number of CPUs the assigned nodes would
     * consume.
     */
    if ((cpu_req == 0) || (cpu_req != (job->nodes * PE_PER_NODE))) {

	/* Compute the "right" memory request, based on the nodes. */
	sprintf(buf, "%d", (job->nodes * PE_PER_NODE));

	if (schd_alterjob(connector, job, ATTR_l, buf, "ncpus")) {
	    DBPRT(("%s: Failed to set job %s \"ncpus\" attr to %s\n", id,
		job->jobid, buf));
	    return (1);
	}

	bumped++;
    }

    if (bumped) {
	strncpy(buf, schd_byte2val(job->nodes * MB_PER_NODE), sizeof(buf) - 1);
	sprintf(log_buffer, "%s cpu/mem (%d/%s) bumped to %d/%s (%d nodes)", 
	    job->jobid, cpu_req, schd_byte2val(mem_req), 
	    job->nodes * PE_PER_NODE, buf, job->nodes);
	log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);
	DBPRT(("%s: %s\n", id, log_buffer));
    }

    return (0);
}

/*
 * The job->priority field correspondes to the priority of
 * the queue to which the job was submitted. This value
 * is then used in the following calculation to determine 
 * a new priority that takes into consideration the recent
 * past usage of the user, the number of nodes requested,
 * the priority of the queue to which the job was submitted,
 * and the lenght of time the job has been queued.
 *
 * Jobs are initially assigned a priority value based on:
 *
 *     Queue     Initial          Max
 *     Name      Priority	Priority
 *     --------  --------       --------
 *     Special     1000	  	  1024
 *     Challenge    900		   999
 *     Normal         0            999
 *     Background    <0          -1024
 *
 */

#define SECS_PER_HOUR 3600

void calc_job_weight(Job *job)
{
    int    timeQd;
    float  usage, weight, A,B,C,D,E,F;
    double priority;
    
    timeQd = schd_TimeNow - job->eligible;
    usage  = get_resource_usage(job->owner); 

    if (!strcmp(job->oqueue, schd_SpecialQueue))
	job->priority = 1000;

    else if (!strcmp(job->oqueue, schd_ChallengeQueue))
	job->priority = 900;

    else if (!strcmp(job->oqueue, schd_BackgroundQueue))
	job->priority = -100;

    else 
	job->priority = 0;


    /* these should be set by parameters in the config file     */
    /* These are used to scale the impact of various attributes */
    /*  of the job on the job weight, used in the priority calc */

    A = 4.0;		/* Adjust recent usage                  */
    B = 2.0;		/* Adjust job size (nodes)              */
    C = 1.0/7.0;	/* Adjust time queued                   */

    /* Use user supplied values for weight of each factor:      */
    /* For simplicty, the user tunable parameters are in the    */
    /* range of 1-10, or 0 (no weight). We scale these into     */
    /* new ranges such that the values proportionately affect   */
    /* the calculated weight of the job.                        */
    /*                                                          */
    /* Recommended input values for a large system are:         */
    /*                                                          */
    /*      schd_USAGE_WEIGHT = 10                              */
    /*      schd_NODES_WEIGHT =  7                              */
    /*      schd_TIMEQ_WEIGHT =  1                              */
    /*                                                          */

    A = schd_USAGE_WEIGHT / 2.50000;      /* scale to range 1-4 */
    B = schd_NODES_WEIGHT / 3.33333;      /* scale to range 1-3 */
    C = schd_TIMEQ_WEIGHT / 1.42857;      /* scale to range 1-7 */

    D = 0.001; 		/* Bring baseline into desired range    */
    E = 0.3;		/* Adjust slope of tanh() curve         */
    F = 1.0;		/* Adjust slope of tanh() curve         */


    /* scale the priority (set by the originating queue) such   */
    /* that the tanh() on the baseline produces appropriate     */
    /* values  (eg. with the ranges stated above).              */

    if      (job->priority >= 1000) job->priority *=  10;
    else if (job->priority >=  900) job->priority *= 7.8;
    else if (job->priority >=    0) job->priority  = 100;
    else if (job->priority <     0) job->priority *=  -1;

    weight   = -A * usage/schd_MAX_NCPUS      +
                B * job->nodes/schd_MAX_NCPUS +
                C * timeQd/SECS_PER_HOUR      +
                D * job->priority             ;

    priority = 500 * ( 1 + tanh( E * weight - F ));
    if (job->priority < 0)
        job->priority = -1 * (1024 - priority);

}