mom_mach.c

openPBS的开放源代码

	while (pres != NULL) {
		assert(pres->rs_defin != NULL);
		pname = pres->rs_defin->rs_name;
		assert(pname != NULL);
		assert(*pname != '\0');

		if (strcmp(pname, "walltime") == 0 ||
		    strcmp(pname, "cput") == 0 ||
		    strcmp(pname, "pcput") == 0 ||
		    strcmp(pname, "vmem") == 0)
			return (TRUE);
		pres = (resource *)GET_NEXT(pres->rs_link);
	}

	return (FALSE);
}

/*
 * Setup for polling.
 *
 * allocate memory for data structure
 */
int mom_open_poll()
{
  return (PBSE_NONE);
}

/*
 * Declare start of polling loop.
 */
int mom_get_sample()
{
  static	char		id[] = "mom_get_sample";
  struct pst_dynamic pst_d;

  DBPRT(("%s: entered\n", id))

  if (pstat_getdynamic(&pst_d, sizeof(struct pst_dynamic), 1, 0) != 1) {
    sprintf(log_buffer, "pstat_getdynamic");
    log_err(errno, id, log_buffer);
    nproc = 0;
    return (PBSE_SYSTEM);
  }
  nproc = pst_d.psd_activeprocs + TBL_INC;
  if (max_proc < nproc) {
    max_proc = nproc;
    proc_status = (struct pst_status *) realloc((void *)proc_status,
                  max_proc*sizeof(struct pst_status));
    if (proc_status == 0) {
      return (PBSE_SYSTEM);
    }
  }
  nproc = pstat_getproc(proc_status, sizeof(struct pst_status), nproc, 0);
  if (nproc < 0) {
    sprintf(log_buffer, "pstat_getproc");
    log_err(errno, id, log_buffer);
    memset(proc_status, sizeof(struct pst_status) * max_proc, '\0');
    nproc = 0;
    return (PBSE_SYSTEM);
  }
  return (PBSE_NONE);
}

/*
 * Measure job resource usage and compare with its limits.
 *
 *	If it has exceeded any well-formed polled limit return TRUE.
 *	Otherwise, return FALSE.
 */

int mom_over_limit(pjob)
    job			*pjob;
{
	char		*id = "mom_over_limit";
	char		*pname;
	int		retval;
	unsigned long	value, num;
	resource	*pres;

	assert(pjob != NULL);
	assert(pjob->ji_wattr[(int)JOB_ATR_resource].at_type == ATR_TYPE_RESC);
	pres = (resource *)
	    GET_NEXT(pjob->ji_wattr[(int)JOB_ATR_resource].at_val.at_list);

	DBPRT(("%s: entered\n", id))
	for ( ; pres != NULL; pres = (resource *)GET_NEXT(pres->rs_link)) {
		assert(pres->rs_defin != NULL);
		pname = pres->rs_defin->rs_name;
		assert(pname != NULL);
		assert(*pname != '\0');
		if (strcmp(pname, "cput") == 0) {
			retval = gettime(pres, &value);
			if (retval != PBSE_NONE)
				continue;
			if ((num = cput_sum(pjob)) > value) {
				sprintf(log_buffer,
					"cput %lu exceeded limit %lu",
					num, value);
				return (TRUE);
			}
		} else if (strcmp(pname, "pcput") == 0) {
			retval = gettime(pres, &value);
			if (retval != PBSE_NONE)
				continue;
			if (overcput_proc(pjob, value)) {
				sprintf(log_buffer,
					"pcput exceeded limit %lu",
					value);
				return (TRUE);
			}
		} else if (strcmp(pname, "vmem") == 0) {
			retval = getsize(pres, &value);
			if (retval != PBSE_NONE)
				continue;
			if ((num = mem_sum(pjob)) > value) {
				sprintf(log_buffer,
					"vmem %lu exceeded limit %lu",
					num, value);
				return (TRUE);
			}
		} else if (strcmp(pname, "walltime") == 0) {
			retval = gettime(pres, &value);
			if (retval != PBSE_NONE)
				continue;
			num = (unsigned long)((double)(time_now - pjob->ji_qs.ji_stime) * wallfactor);
			if (num > value) {
				sprintf(log_buffer,
					"walltime %d exceeded limit %d",
					num, value);
				return (TRUE);
			}
		}
	}

	return (FALSE);
}

/*
 * Update the job attribute for resources used.
 *
 *	The first time this is called for a job, set up resource entries for
 *	each resource that can be reported for this machine.  Fill in the
 *	correct values.  Return an error code.
 *
 *	Assumes that the session ID attribute has already been set.
 */
int mom_set_use(pjob)
    job			*pjob;
{
	char			*id = "mom_set_use";
	resource		*pres;
	attribute		*at;
	resource_def		*rd;
	unsigned long		*lp, lnum;

	assert(pjob != NULL);
	at = &pjob->ji_wattr[(int)JOB_ATR_resc_used];
	assert(at->at_type == ATR_TYPE_RESC);

	at->at_flags |= ATR_VFLAG_MODIFY;
	if ((at->at_flags & ATR_VFLAG_SET) == 0) {
		at->at_flags |= ATR_VFLAG_SET;

		rd = find_resc_def(svr_resc_def, "cput", svr_resc_size);
		assert(rd != NULL);
		pres = add_resource_entry(at, rd);
		pres->rs_value.at_flags |= ATR_VFLAG_SET;
		pres->rs_value.at_type = ATR_TYPE_LONG;

		rd = find_resc_def(svr_resc_def, "vmem", svr_resc_size);
		assert(rd != NULL);
		pres = add_resource_entry(at, rd);
		pres->rs_value.at_flags |= ATR_VFLAG_SET;
		pres->rs_value.at_type = ATR_TYPE_SIZE;
		pres->rs_value.at_val.at_size.atsv_shift = 10; /* KB */
		pres->rs_value.at_val.at_size.atsv_units = ATR_SV_BYTESZ;

		rd = find_resc_def(svr_resc_def, "walltime", svr_resc_size);
		assert(rd != NULL);
		pres = add_resource_entry(at, rd);
		pres->rs_value.at_flags |= ATR_VFLAG_SET;
		pres->rs_value.at_type = ATR_TYPE_LONG;
	}

	rd = find_resc_def(svr_resc_def, "cput", svr_resc_size);
	assert(rd != NULL);
	pres = find_resc_entry(at, rd);
	assert(pres != NULL);
	lp = (unsigned long *)&pres->rs_value.at_val.at_long;
	lnum = cput_sum(pjob);
	*lp = MAX(*lp, lnum);

	rd = find_resc_def(svr_resc_def, "vmem", svr_resc_size);
	assert(rd != NULL);
	pres = find_resc_entry(at, rd);
	assert(pres != NULL);
	lp = &pres->rs_value.at_val.at_size.atsv_num;
	lnum = (mem_sum(pjob) + 1023) >> 10; /* KB */
	*lp = MAX(*lp, lnum);

	rd = find_resc_def(svr_resc_def, "walltime", svr_resc_size);
	assert(rd != NULL);
	pres = find_resc_entry(at, rd);
	assert(pres != NULL);
	pres->rs_value.at_val.at_long = (long)((double)(time_now - pjob->ji_qs.ji_stime) * wallfactor);

	return (PBSE_NONE);
}

/*
 *	Kill a task session.
 *	Call with the task pointer and a signal number.
 */
int kill_task(ptask, sig)
    task	*ptask;
    int  	sig;
{
	char		        *id = "kill_task";
	int		        ct = 0;
	struct pst_status	*ps;
	int		        i, sesid;

	sesid = ptask->ti_qs.ti_sid;
	if (sesid <= 1)
		return 0;

	(void)mom_get_sample();

	for (i=0; i<nproc; i++) {
		ps = &proc_status[i];

		if (sesid == ps->pst_sid) {
			(void)kill(ps->pst_pid, sig);
			++ct;
		}
	}
	return ct;
}

/*
 * Clean up everything related to polling.
 *
 */
int mom_close_poll()
{
	char	*id = "mom_close_poll";

	DBPRT(("%s: entered\n", id))
	if (proc_status)
		free(proc_status);

	return (PBSE_NONE);
}

/*
 * mom_does_chkpnt - return 1 if mom supports checkpoint
 *			    0 if not
 */

int mom_does_chkpnt()
{
	return (0);
}

/*
 * Checkpoint the job.
 *
 *	If abort is true, kill it too.  Return -1 on error.
 */

int mach_checkpoint(ptask, file, abort)
    task	*ptask;
    char	*file;
    int		abort;
{
       	return (-1);
}

/*
 * Restart the job from the checkpoint file.
 */

long mach_restart(ptask, file)
    task	*ptask;
    char	*file;
{
	return (-1);
}

/*
**	Return 1 if proc table can be read, 0 otherwise.
*/
int
getprocs()
{
	static	unsigned	int	lastproc = 0;
	static	char		id[] = "getprocs";

	if (lastproc == reqnum)		/* don't need new proc table */
		return 1;

	if (mom_get_sample() != PBSE_NONE)
		return 0;

	lastproc = reqnum;
	return 1;
}

char	*
cput_job(jobid)
pid_t	jobid;
{
	char			*id = "cput_job";
	int			found = 0;
	int			i;
	double			cputime, addtime;
	struct pst_status	*ps;

	if (getprocs() == 0) {
		rm_errno = RM_ERR_SYSTEM;
		return NULL;
	}

	cputime = 0.0;
	for (i=0; i<nproc; i++) {
		ps = &proc_status[i];

		if (jobid != ps->pst_sid)
			continue;

		found = 1;
		addtime = ps->pst_utime +
			ps->pst_stime;
                        /*
			ps->pr_cutime +
			ps->pr_cstime;
                        */

		cputime += addtime;
		DBPRT(("%s: total %.2f pid %d %.2f\n", id, cputime,
				ps->pst_pid, addtime))

	}
	if (found) {
		sprintf(ret_string, "%.2f", cputime * cputfactor);
		return ret_string;
	}

	rm_errno = RM_ERR_EXIST;
	return NULL;
}

char	*
cput_proc(pid)
pid_t	pid;
{
	char			*id = "cput_pid";
	double			cputime;
	int			i;
	struct pst_status	*ps;

	if (getprocs() == 0) {
		rm_errno = RM_ERR_SYSTEM;
		return NULL;
	}

	for (i=0; i<nproc; i++) {
		ps = &proc_status[i];

		if (ps->pst_pid == pid)
			break;
	}
	if (i == nproc) {
		rm_errno = RM_ERR_EXIST;
		return NULL;
	}
	cputime = ps->pst_utime +
		ps->pst_stime;
                /*
		ps->pr_cutime +
		ps->pr_cstime;
                */

	sprintf(ret_string, "%.2f", cputime * cputfactor);
	return ret_string;
}


char	*
cput(attrib)
struct	rm_attribute	*attrib;
{
	char			*id = "cput";
	int			value;

	if (attrib == NULL) {
		log_err(-1, id, no_parm);
		rm_errno = RM_ERR_NOPARAM;
		return NULL;
	}
	if ((value = atoi(attrib->a_value)) == 0) {
		sprintf(log_buffer, "bad param: %s", attrib->a_value);
		log_err(-1, id, log_buffer);
		rm_errno = RM_ERR_BADPARAM;
		return NULL;
	}
	if (momgetattr(NULL)) {
		log_err(-1, id, extra_parm);
		rm_errno = RM_ERR_BADPARAM;
		return NULL;
	}

	if (strcmp(attrib->a_qualifier, "session") == 0)
		return (cput_job((pid_t)value));
	else if (strcmp(attrib->a_qualifier, "proc") == 0)
		return (cput_proc((pid_t)value));
	else {
		rm_errno = RM_ERR_BADPARAM;
		return NULL;
	}
}

char	*
mem_job(jobid)
pid_t	jobid;
{
	char			*id = "mem_job";
	int			memsize;
	int			i;
	int			found = 0;
	struct pst_status	*ps;

	if (getprocs() == 0) {
		rm_errno = RM_ERR_SYSTEM;
		return NULL;
	}

	memsize = 0;
	for (i=0; i<nproc; i++) {
		ps = &proc_status[i];

		if (jobid != ps->pst_sid)
			continue;

		found = 1;
		memsize += VSIZE(ps);
		DBPRT(("%s: total %d pid %d %d\n", id, memsize*page_size,
				ps->pst_pid, VSIZE(ps)*page_size))
	}
	if (found) {
		/* in KB */
		sprintf(ret_string, "%ukb", (memsize * page_size) >> 10);
		return ret_string;
	}

	rm_errno = RM_ERR_EXIST;
	return NULL;
}

char	*
mem_proc(pid)
pid_t	pid;
{
	char			*id = "mem_proc";
	struct pst_status	*ps;
	int			i;

	if (getprocs() == 0) {
		rm_errno = RM_ERR_SYSTEM;
		return NULL;
	}

	for (i=0; i<nproc; i++) {
		ps = &proc_status[i];

		if (ps->pst_pid == pid)
			break;
	}
	if (i == nproc) {
		rm_errno = RM_ERR_EXIST;
		return NULL;
	}

	sprintf(ret_string, "%ukb", (VSIZE(ps) * page_size) >> 10); /* KB */
	return ret_string;
}

char	*
mem(attrib)
struct	rm_attribute	*attrib;
{
	char			*id = "mem";
	int			value;

	if (attrib == NULL) {
		log_err(-1, id, no_parm);
		rm_errno = RM_ERR_NOPARAM;
		return NULL;
	}
	if ((value = atoi(attrib->a_value)) == 0) {
		sprintf(log_buffer, "bad param: %s", attrib->a_value);
		log_err(-1, id, log_buffer);
		rm_errno = RM_ERR_BADPARAM;
		return NULL;
	}
	if (momgetattr(NULL)) {
		log_err(-1, id, extra_parm);
		rm_errno = RM_ERR_BADPARAM;
		return NULL;
	}

	if (strcmp(attrib->a_qualifier, "session") == 0)
		return (mem_job((pid_t)value));
	else if (strcmp(attrib->a_qualifier, "proc") == 0)
		return (mem_proc((pid_t)value));
	else {
		rm_errno = RM_ERR_BADPARAM;
		return NULL;
	}
}

static char	*
resi_job(jobid)
pid_t	jobid;
{
	char			*id = "resi_job";
	int			resisize;
	int			i;
	int			found = 0;
	struct pst_status	*ps;

	if (getprocs() == 0) {
		rm_errno = RM_ERR_SYSTEM;
		return NULL;
	}

	resisize = 0;
	for (i=0; i<nproc; i++) {
		ps = &proc_status[i];

		if (jobid != ps->pst_sid)
			continue;

		found = 1;
		resisize += ps->pst_rssize;
	}
	if (found) {
		sprintf(ret_string, "%ukb",(resisize*page_size) >> 10); /* KB */
		return ret_string;
	}

	rm_errno = RM_ERR_EXIST;
	return NULL;
}

static char	*
resi_proc(pid)
pid_t	pid;
{
	char			*id = "resi_proc";
	struct pst_status	*ps;
	int			i;

	if (getprocs() == 0) {
		rm_errno = RM_ERR_SYSTEM;
		return NULL;
	}

	for (i=0; i<nproc; i++) {
		ps = &proc_status[i];

		if (ps->pst_pid == pid)
			break;
	}
	if (i == nproc) {
		rm_errno = RM_ERR_EXIST;
		return NULL;
	}

	sprintf(ret_string, "%ukb", (ps->pst_rssize*page_size) >> 10); /* KB */
	return ret_string;
}

static char	*
resi(attrib)
struct	rm_attribute	*attrib;
{
	char			*id = "resi";
	int			value;

	if (attrib == NULL) {
		log_err(-1, id, no_parm);
		rm_errno = RM_ERR_NOPARAM;
		return NULL;
	}
	if ((value = atoi(attrib->a_value)) == 0) {
		sprintf(log_buffer, "bad param: %s", attrib->a_value);
		log_err(-1, id, log_buffer);
		rm_errno = RM_ERR_BADPARAM;
		return NULL;
	}
	if (momgetattr(NULL)) {
		log_err(-1, id, extra_parm);
		rm_errno = RM_ERR_BADPARAM;
		return NULL;
	}