usersort.c

openPBS的开放源代码

 * other than at the end to free all the sublists.
 *
 * The final list will be reassembled and ordered as: 
 *
 *      running, normal
 */
static Job *
make_job_list(void)
{
    Job    list_seed, *joblist, *jobtail;
    int    i;

    memset(&list_seed, 0, sizeof(list_seed));

    /*
     * "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);
}

/*
 * Construct a list of users who own one or more "normal" jobs
 */
void get_users(void)
{
    char   *id = "get_users";
    Job    *job_ptr;
    char   *uname, *name;
    char   que_usr_tuple[120]; 
    int     which, i, j, max, unused;
    static int is_new_user();
    static int make_uinfo();

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

    Users  = NULL;
    nUsers = 0;

    /* 
     * Walk the list of "normal" jobs, adding any new users along the way.
     */
    for (i = 0; i < nnormalQ; ++i) {

	job_ptr = normalQ[i];
	sprintf(que_usr_tuple, "%s:%s", job_ptr->owner, job_ptr->oqueue);

	/* Is this a new entry in the list? */
	if (is_new_user(que_usr_tuple, Users, nUsers, &which)) {

	    ++nUsers;

	    Users = realloc(Users, nUsers * sizeof *Users);
	    if (!Users) {
		log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER,
			   id, "realloc(Users)");
		return;
	    }
	    make_uinfo(que_usr_tuple, &(Users[nUsers - 1]));
	} else {
	    Users[which].jobcount++;
	}
    }

    /* Now, walk the list of running jobs and record each user's count. */
    for (i = 0; i < nUsers; ++i) {
	uname = Users[i].name;
	for (j = 0; j < nJRs; ++j) {
	    job_ptr = runningJobs[j];
	    sprintf(que_usr_tuple, "%s:%s", job_ptr->owner, job_ptr->oqueue);

	    if (!strcmp(que_usr_tuple, uname)) {
		Users[i].running_jobs   ++;
	        schd_accesslimits(job_ptr->oqueue, &max, &unused);
		if (Users[i].running_jobs > max)
		    job_ptr->flags |= JFLAGS_RUNLIMIT;
	    }
	    ++job_ptr;
	}
    }
    return;
}

/* Search the list of users, looking for 'user'. */
static int
is_new_user(char *user, struct Uinfo *list, int len, int *which)
{
    struct Uinfo *list_ptr = list;
    int     i;

    for (i = 0; i < len; ++i) {
	if (!strcmp(list_ptr->name, user)) {
	    *which = i;
	    return 0;
	}
	++list_ptr;
    }

    return 1;
}

/* Build a record of the pertinent data about a job owner. */
static int
make_uinfo(char *user, struct Uinfo *uinfo)
{
    strncpy(uinfo->name, user, sizeof(uinfo->name) - 1);
    uinfo->running_jobs   = 0;
    uinfo->jobcount       = 1;

    return 0;
}

/* Determine if it is time to save the Share Usage information to disk */
int
schd_save_shares(void)
{
    					/* save shares hourly */
    if (schd_TimeNow >= (when_shares_saved + 3600))
	return 1;
    return 0;
}

/* 
 * Read or write the Share Usage data to the disk file.
 */
void
schd_share_info(char *mode)
{
    char   *id = "schd_share_info";
    FairAccessList *Fptr;
    AccessEntry    *AEptr;
    FILE   *shares;
    char   *p;
    char    buffer[MAX_TXT + 1 + 1]; /* size = MAX_TXT + a newline + NULL */
    char    qname[50], share_today[10], share_day[10];
    int     do_shares = 0;
    int	    past_ndays;
    double  past_percent, today_usage, today_max;


    /* get the current day of the week (name) */
    strftime(share_today, sizeof share_today, "%a", &schd_TmNow);

    /* If there is no pre-existing file, bootstrap it with the date */
    if ((shares = fopen(schd_SHARE_INFOFILE, mode)) == NULL) {

	if ((shares = fopen(schd_SHARE_INFOFILE, "w")) == NULL) {
	    /*
	     * Could not open the file for writing.  Possibly the path is
	     * invalid?
	     */
	    sprintf(log_buffer, "fopen(%s,%s) failed (%d)",
		    schd_SHARE_INFOFILE, mode, errno);
	    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);

	    return;
	}

	/* 
	 * The file pointed to by schd_SHARE_INFOFILE is writable.  Create the
	 * bootstrap record.
	 */

	sprintf(buffer, "# Day-of-Week\n# queue  ndays past%% today_usage today_max\n");
	fputs(buffer, shares);

	sprintf(buffer, "%s\n", share_today);
	fputs(buffer, shares);

	fclose(shares);

	sprintf(log_buffer, "wrote bootstrap record");
	log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);

	/* 
	 * Attempt to reopen the file.  If it fails, then the bootstrap failed
 	 * as well.  Otherwise, this leaves the open stream pointer in 'shares'
 	 * as if it were opened successfully above.
	 */
	if ((shares = fopen(schd_SHARE_INFOFILE, mode)) == NULL) {
	    sprintf(log_buffer, "bootstrap:fopen(%s,%s) failed",
		    schd_SHARE_INFOFILE, mode);
	    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);
	    return;
	}
    }

    /* Read or write based on mode. */
    if (mode[0] == 'r') {

	fgets(buffer, sizeof buffer, shares);	/* skip both header comments */
	fgets(buffer, sizeof buffer, shares);

	fgets(buffer, sizeof buffer, shares);	/* save Day-of-week stamp */
	p = strtok(buffer, " \n");
	strcpy(share_day, p);

	while (fgets(buffer, sizeof buffer, shares)) {

	    p = strtok(buffer, " \n");
	    strcpy(qname, p);

	    p = strtok(NULL, " \n");
	    past_ndays = atoi(p);

	    p = strtok(NULL, " \n");
	    past_percent = atof(p);

	    p = strtok(NULL, " \n");
	    today_usage = atof(p);

	    p = strtok(NULL, " \n");
	    today_max = atof(p);

	    
	    /* now save this info in the global per-Queue Shares lists */
            for (Fptr = schd_FairACL; Fptr != NULL; Fptr = Fptr->next) {
                for (AEptr = Fptr->entry; AEptr != NULL; AEptr = AEptr->next) {
                    if (AEptr->name) {
                        if (!strcmp(AEptr->name, qname)) {
	                    AEptr->past_ndays   = past_ndays;
	                    AEptr->past_percent = past_percent;
	                    AEptr->today_usage  = today_usage;
	                    AEptr->today_max    = today_max;

			    if(strcmp(share_today, share_day)) {
	    		        /* If Day is not Today, then we need to   */
				/* apply the "today" usage data to the    */
				/* "past", and clear the today counters;  */
				/* this is needed in the off chance the   */
				/* scheduler stopped between the hourly   */
				/* updates and the daily today-past usage */
				/* conversion. We don't want to loose any */
				/* Share usage data.		 	  */

				update_share_usage(AEptr);
			    }
	    		    sprintf(log_buffer,
			        "%10s: %d shares; averaged %3.1f shares "
				"over past %d days", qname, AEptr->max_percent,
				AEptr->past_percent, AEptr->past_ndays);
	    		    log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER,
				id,log_buffer);
	                }
                    }
                }
            }
	}
	fclose(shares);

	schd_NeedToGetShareInfo = 0;
	when_shares_done = next_shares_update();

    } else if (mode[0] == 'w') {

	do_shares = schd_TimeNow > when_shares_done;

	sprintf(buffer, "# Day-of-Week\n# queue ndays past%% today_usage today_max\n");
	fputs(buffer, shares);
	sprintf(buffer, "%s\n", share_today);
	fputs(buffer, shares);

        for (Fptr = schd_FairACL; Fptr != NULL; Fptr = Fptr->next) {
            for (AEptr = Fptr->entry; AEptr != NULL; AEptr = AEptr->next) {
                if (AEptr->name) {
	            if (do_shares) {

			/* convert today usage into past usage */
			update_share_usage(AEptr);

	    	        sprintf(log_buffer,
		            "%10s: %d shares; averaged %3.1f shares over past %d days",
			    AEptr->name, AEptr->max_percent, AEptr->past_percent,
			    AEptr->past_ndays);
	    	        log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id,
			    log_buffer);
	            }

		    /* now write out the current values */
	    	    fprintf(shares, "%-10s %5d %3.3f %10.3f %10.3f\n",
			AEptr->name, AEptr->past_ndays, AEptr->past_percent,
			AEptr->today_usage, AEptr->today_max);
		}
	    }
	}
	fclose(shares);
	if (do_shares)
	    when_shares_done = next_shares_update();

	when_shares_saved = schd_TimeNow;

    } else {
	sprintf(log_buffer, "unknown mode [%s]", mode);
	log_record(PBSEVENT_SYSTEM, PBS_EVENTCLASS_SERVER, id, log_buffer);
    }

    return;
}

static time_t
next_shares_update(void)
{
    struct tm *tmptr;
    time_t  when, next, secSinceMidnite;

    /* Calculate number of seconds elapsed since midnight (00:00:00). */
    time(&when);
    tmptr = localtime(&when);
    secSinceMidnite = ((tmptr->tm_hour * 60 * 60 ) +
                      ( tmptr->tm_min  * 60 )      +
		        tmptr->tm_sec );

	/* schedule next update at 23:00 tonight */
    next = schd_TimeNow - secSinceMidnite + 60*60*23;
		
    return(next);
}

static void
count_user_jobs(Job *job, int *running, int *total, int *max)
{
    char   que_usr_tuple[120];
    int i, unused;
    *running = 0;
    *total   = 0;
    *max     = 1;

    sprintf(que_usr_tuple, "%s:%s", job->owner, job->oqueue);

    for (i = 0; i < nUsers; ++i) {
	if (!strcmp(que_usr_tuple, Users[i].name)) {
	    *running = Users[i].running_jobs;
	    *total   = Users[i].jobcount;
	    schd_accesslimits(job->oqueue, max, &unused);
	    break;
        }
    }
}

static void mark_overusage_jobs(void)
{
    char   *id = "mark_overusage_jobs";
    int i, running, total, max;
    Job    *job_ptr;

    /* Walk running jobs list marking any jobs that are possible candidates
     * for suspension/checkpointing. We do this here so that *if* we need
     * find a job to checkpoint later, we will have already have the list
     * to choose from.
     */
    for (i = 0; i < nJRs; i++) {
	
	job_ptr = runningJobs[i];

	/* skip Priority or Long Waiting jobs */
	if (job_ptr->flags & JFLAGS_PRIORITY ||
	    job_ptr->flags & JFLAGS_WAITING )
	    continue;

	/* check if share usage exceeds 100 % */
	if (job_ptr->sort_order > 100) {
	    job_ptr->flags |= JFLAGS_CHKPT_OK;
	    continue;
	}

	/* can add other checks here */
    }

    /* now mark any Q'd jobs from overusage queue/etc. */
    for (i = 0; i < nnormalQ; i++) {
	job_ptr = normalQ[i];

	/* check if user has exceeded max running jobs limit */
	count_user_jobs(job_ptr, &running, &total, &max);
	if (running > max) {
	    job_ptr->flags |= JFLAGS_RUNLIMIT;
	    continue;
	}

	/* check if share usage exceeds 100 % */
	if (job_ptr->sort_order > 100) {
	    job_ptr->flags |= JFLAGS_CHKPT_OK;
	    continue;
	}
    }
}