mom_mach.c

openPBS的开放源代码

{
	ulong		memsize;
	int		i;
	prpsinfo_t	*pi;

	memsize = 0;
	for (i=0; i<nproc; i++) {
		pi = &proc_array[i].info;

		if (!injob(phead, pi))
			continue;
                if ((rlim64_t)((rlim64_t)pagesize * pi->pr_size) > limit)
			return (TRUE);
        }

	return (FALSE);
}

extern char *msg_momsetlim;

/*
 * Internal error routine
 */
int error(string, value)
    char	*string;
    int		value;
{
	int		i = 0;
	char		*message;

	assert(string != NULL);
	assert(*string != '\0');
	assert(value > PBSE_);			/* minimum PBS error number */
	assert(value <= PBSE_NOSYNCMSTR);	/* maximum PBS error number */
	assert(pbs_err_to_txt[i].err_no != 0);

	do {
		if (pbs_err_to_txt[i].err_no == value)
			break;
	} while (pbs_err_to_txt[++i].err_no != 0);

	assert(pbs_err_to_txt[i].err_txt != NULL);
	message = *pbs_err_to_txt[i].err_txt;
	assert(message != NULL);
	assert(*message != '\0');
	(void)fprintf(stderr, msg_momsetlim, string, message);
	(void)fflush(stderr);

	return (value);
}

/*
 * Establish system-enforced limits for the job.
 *
 *	Run through the resource list, checking the values for all items
 *	we recognize.
 *
 *	If set_mode is SET_LIMIT_SET, then also set hard limits for the
 *	system enforced limits (not-polled).
 *	If anything goes wrong with the process, return a PBS error code
 *	and print a message on standard error.  A zero-length resource list
 *	is not an error.
 *
 *	If set_mode is SET_LIMIT_SET the entry conditions are:
 *	    1.	MOM has already forked, and we are called from the child.
 *	    2.	The child is still running as root.
 *	    3.  Standard error is open to the user's file.
 *
 *	If set_mode is SET_LIMIT_ALTER, we are beening called to modify
 *	existing limits.  Cannot alter those set by setrlimit (kernel)
 *	because we are the wrong process.  
 */
int mom_set_limits(pjob, set_mode)
    job			*pjob;
    int			 set_mode;	/* SET_LIMIT_SET or SET_LIMIT_ALTER */
{
	char		*id = "mom_set_limits";
	char		*pname;
	int		retval;
	rlim64_t	sizeval; /* place to build 64 bit value */
	unsigned long	value;	/* place in which to build resource value */
	resource	*pres;
       	struct rlimit64	res64lim;
	rlim64_t	mem_limit  = 0;
#if	NODEMASK != 0
	__uint64_t	rvalue;
	__uint64_t	nodemask;
#endif	/* NODEMASK */

	DBPRT(("%s: entered\n", id))
	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);

/*
 * Cycle through all the resource specifications,
 * setting limits appropriately.
 */

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

		if (strcmp(pname, "ncpus") == 0) {
			char	hold[16];
			extern	struct	var_table	vtable;

			retval = getlong(pres, &value);
			if (retval != PBSE_NONE)
			        return (error(pname, retval));
			sprintf(hold, "%d", (int)pres->rs_value.at_val.at_long);
			bld_env_variables(&vtable, "NCPUS", hold);
		} else if (strcmp(pname, "cput") == 0) {	/* check */
			/* cpu time - check, if less than pcput use it */
			retval = gettime(pres, &value);
			if (retval != PBSE_NONE)
			        return (error(pname, retval));
		} else if (strcmp(pname, "pcput") == 0) {
			/* process cpu time - set */
			retval = gettime(pres, &value);
			if (retval != PBSE_NONE)
			        return (error(pname, retval));
			res64lim.rlim_cur = res64lim.rlim_max =
                                (unsigned long)((double)value / cputfactor);
			if (setrlimit64(RLIMIT_CPU, &res64lim) < 0)
	        		return (error("RLIMIT_CPU", PBSE_SYSTEM));
		} else if (strcmp(pname, "file") == 0) {	/* set */
			if (set_mode == SET_LIMIT_SET)  {
			    retval = getsize(pres, &sizeval);
			    if (retval != PBSE_NONE)
			        return (error(pname, retval));
			    res64lim.rlim_cur = res64lim.rlim_max = sizeval;
			    if (setrlimit64(RLIMIT_FSIZE, &res64lim) < 0)
			        return (error(pname, PBSE_SYSTEM));
			}
		} else if (strcmp(pname, "vmem") == 0) {	/* check */
			retval = getsize(pres, &sizeval);
			if (retval != PBSE_NONE)
			        return (error(pname, retval));
			if ((mem_limit == 0) || (sizeval < mem_limit))
				mem_limit = sizeval;
		} else if (strcmp(pname, "pvmem") == 0) {	/* set */
			retval = getsize(pres, &sizeval);
			if (retval != PBSE_NONE)
			        return (error(pname, retval));
			if ((mem_limit == 0) || (sizeval < mem_limit))
				mem_limit = sizeval;
		} else if (strcmp(pname, "mem") == 0) {		/* ignore */
		} else if (strcmp(pname, "pmem") == 0) {	/* set */
			if (set_mode == SET_LIMIT_SET)  {
			    retval = getsize(pres, &sizeval);
			    if (retval != PBSE_NONE)
			        return (error(pname, retval));
			    res64lim.rlim_cur = res64lim.rlim_max = sizeval;
			    if (setrlimit64(RLIMIT_RSS, &res64lim) < 0)
	        		return (error("RLIMIT_RSS", PBSE_SYSTEM));
			}
		} else if (strcmp(pname, "walltime") == 0) {	/* Check */
			retval = getlong(pres, &value);
			if (retval != PBSE_NONE)
			        return (error(pname, retval));
		} else if (strcmp(pname, "nice") == 0) {	/* set nice */
			if (set_mode == SET_LIMIT_SET)  {
			    errno = 0;
			    if ((nice((int)pres->rs_value.at_val.at_long) == -1)
				&& (errno != 0))
				return (error(pname, PBSE_BADATVAL));
			}
#if	NODEMASK != 0
		} else if (strcmp(pname, "nodemask") == 0) {  /* set nodemask */
			/* call special node mask function */
			nodemask = pres->rs_value.at_val.at_ll;
			rvalue = (__uint64_t)pmoctl(61, nodemask,0);
			if (rvalue != nodemask) {
				(void)sprintf(log_buffer, "Tried to set node mask to 0x%0llx, was set to 0x%0llx", nodemask, rvalue);
				LOG_EVENT(PBSEVENT_ERROR, PBS_EVENTCLASS_JOB,
					  pjob->ji_qs.ji_jobid, log_buffer);
			}
#endif	/* NODEMASK */
		} else if ((pres->rs_defin->rs_flags & ATR_DFLAG_RMOMIG) == 0)
			/* don't recognize and not marked as ignore by mom */
			return (error(pname, PBSE_UNKRESC));
		pres = (resource *)GET_NEXT(pres->rs_link);
	}
	if (set_mode == SET_LIMIT_SET)  {
	    /* if either of vmem or pvmem was given, set sys limit to lesser */
	    if (mem_limit != 0) {
		res64lim.rlim_cur = res64lim.rlim_max = mem_limit;
		if (setrlimit64(RLIMIT_VMEM, &res64lim) < 0)
			return (error("RLIMIT_VMEM", PBSE_SYSTEM));
	    }
	}
	return (PBSE_NONE);
}
/*
 * State whether MOM main loop has to poll this job to determine if some
 * limits are being exceeded.
 *
 *	Sets flag TRUE if polling is necessary, FALSE otherwise.  Actual
 *	polling is done using the mom_over_limit machine-dependent function.
 */

int mom_do_poll(pjob)
    job			*pjob;
{
	char		*id = "mom_do_poll";
	char		*pname;
	resource	*pres;

	DBPRT(("%s: entered\n", id))
	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);

	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, "ncpus") == 0 ||
		    strcmp(pname, "cput") == 0 ||
		    strcmp(pname, "pcput") == 0 ||
		    strcmp(pname, "pvmem") == 0 ||
		    strcmp(pname, "vmem") == 0)
			return (TRUE);
		pres = (resource *)GET_NEXT(pres->rs_link);
	}

	return (FALSE);
}

/*
 * Setup for polling.
 *
 *	Open kernel device and get namelist info.
 *	Also open sgi project files.
 */
int mom_open_poll()
{
	char		*id = "mom_open_poll";
	extern int	 open_sgi_proj();

	DBPRT(("%s: entered\n", id))
	pagesize = getpagesize();

	proc_array = (struct proc_info *)calloc(TBL_INC,
				sizeof(struct proc_info));
	if (proc_array == NULL) {
		log_err(errno, id, "malloc");
		return (PBSE_SYSTEM);
	}
	max_proc = TBL_INC;

	return (open_sgi_proj());
}

/*
 * Declare start of polling loop.
 */

int mom_get_sample()
{
	static	char		id[] = "mom_get_sample";
	int			fd;
	struct dirent		*dent;
	char			procname[100];
	int			num;
	int			mapsize;
	time_t			currtime;
	prmap_sgi_arg_t		maparg;
	struct	proc_info	*pi;
	prmap_sgi_t		map[MAPNUM];

	DBPRT(("%s: entered pagesize %d\n", id, pagesize))
	rewinddir(pdir);
	nproc = 0;
	pi = proc_array;

	mapsize = sizeof(prmap_sgi_t) * MAPNUM;
	maparg.pr_size = mapsize;

	currtime = time(0);
	for (fd = -1; (dent = readdir(pdir)) != NULL; close(fd)) {
		if (!isdigit(dent->d_name[0]))
			continue;

		sprintf(procname, procfmts, dent->d_name);
		if ((fd = open(procname, O_RDONLY)) == -1)
			continue;

		if (ioctl(fd, PIOCPSINFO, &pi->info) == -1) {
			if (errno != ENOENT) {
				sprintf(log_buffer,
					"%s: ioctl(PIOCPSINFO)", procname);
				log_err(errno, id, log_buffer);
			}
			continue;
		}

#if COMPLEX_MEM_CALC==1
		if (pi->map) {
			free(pi->map);		/* free any old space */
			pi->map = NULL;
		}
		pi->map_num = 0;
		maparg.pr_vaddr = (caddr_t)map;

		if ((num = ioctl(fd, PIOCMAP_SGI, &maparg)) == -1) {
			if (errno != ENOENT)
				log_err(errno, id, "ioctl(PIOCMAP_SGI)");
			continue;
		}
		if (num > 0) {
			size_t	nb = sizeof(prmap_sgi_t) * num;

			assert(num < MAPNUM);
			pi->map = (prmap_sgi_t *) malloc(nb);
			memcpy(pi->map, map, nb);
			pi->map_num = num;
		}
#endif	/* COMPLEX_MEM_CALC */

		if (++nproc == max_proc) {
			struct	proc_info	*hold;

			DBPRT(("%s: alloc more table space %d\n", id, nproc))
			max_proc *= 2;

			hold = (struct proc_info *)realloc(proc_array,
					max_proc*sizeof(struct proc_info));
			assert(hold != NULL);
			proc_array = hold;
			memset(&proc_array[nproc], '\0',
				sizeof(struct proc_info) * (max_proc >> 1));
		}
		pi = &proc_array[nproc];
	}
	sampletime = time(0);

	if ((sampletime - currtime) > 5) {
		sprintf(log_buffer, "time lag %ld secs", (long)(sampletime-currtime));
		log_err(-1, id, log_buffer);
		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		*pname;
	int		retval;
	rlim64_t	sizeval;
	unsigned long	value, num;
	rlim64_t	num64;
	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);

	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, "ncpus") == 0) {
			attribute		*at;
			resource_def		*rd;
			resource		*prescpup;

			retval = getlong(pres, &value);
			if (retval != PBSE_NONE)
				continue;

			at = &pjob->ji_wattr[(int)JOB_ATR_resc_used];
			assert(at->at_type == ATR_TYPE_RESC);
			rd = find_resc_def(svr_resc_def, "cpupercent",
				svr_resc_size);
			assert(rd != NULL);
			prescpup = find_resc_entry(at, rd);
			assert(prescpup != NULL);

			num = prescpup->rs_value.at_val.at_long;
			if (num > (value*100+10)) {
				sprintf(log_buffer,
					"ncpus %.2f exceeded limit %lu",
					(float)num/100.0, value);
#if !defined(SGI_ZOMBIE_WRONG)
				return (TRUE);
#else
				LOG_EVENT(PBSEVENT_JOB, PBS_EVENTCLASS_JOB,
					  pjob->ji_qs.ji_jobid, log_buffer);
#endif /* SGI_ZOMBIE_WRONG */
			}
		} else if (strcmp(pname, "cput") == 0) {
			retval = getlong(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 = getlong(pres, &value);
			if (retval != PBSE_NONE)
				continue;
			if (overcpu_proc(&pjob->ji_tasks, value)) {
				sprintf(log_buffer,
					"pcput exceeded limit %lu", value);
				return (TRUE);
			}
		} else if (strcmp(pname, "vmem") == 0) {
			retval = getsize(pres, &sizeval);
			if (retval != PBSE_NONE)
				continue;
			if ((num64 = mem_sum(&pjob->ji_tasks)) > sizeval) {
				sprintf(log_buffer,
					"vmem %llu exceeded limit %llu",
					num64, sizeval);
				return (TRUE);
			}
		} else if (strcmp(pname, "pvmem") == 0) {
			retval = getsize(pres, &sizeval);
			if (retval != PBSE_NONE)
				continue;
			if (overmem_proc(&pjob->ji_tasks, sizeval)) {
				sprintf(log_buffer,
					"pvmem exceeded limit %llukb", sizeval);
				return (TRUE);
			}
		} else if (strcmp(pname, "walltime") == 0) {
			retval = getlong(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 %ld exceeded limit %lu",
					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.
 */
int mom_set_use(pjob)
    job			*pjob;
{
	resource		*pres;
	attribute		*at;
	resource_def		*rd;
	unsigned long		*lp, lnum, newcpu, oldcpu; 
	long			 dur;
	unsigned long  		 percent;

	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;
		pres->rs_value.at_val.at_long = 0;

		rd = find_resc_def(svr_resc_def, "cpupercent", 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;
		pres->rs_value.at_val.at_long = 0;

		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; /* in 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, "mem", 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; /* in KB */
		pres->rs_value.at_val.at_size.atsv_units = ATR_SV_BYTESZ;
	}

	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;
	oldcpu = *lp;
	lnum = cput_sum(pjob);
	if (lnum > *lp) {
		*lp = lnum;

		if ( (dur = sampletime - pjob->ji_sampletim) > 10) {
			newcpu = *lp;
			rd = find_resc_def(svr_resc_def, "cpupercent", 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;
			percent = (newcpu - oldcpu)*100 / dur;
			*lp = MAX(*lp, percent);
			DBPRT(("cpu %%  : ses %ld (new %lu - old %lu)/delta %ld = %lu%%\n", pjob->ji_wattr[(int)JOB_ATR_session_id].at_val.at_long, newcpu, oldcpu, dur, percent))
		}
		pjob->ji_sampletim = sampletime;
	}

	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->ji_tasks) + 1023) >> 10;	/* as 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);

	rd = find_resc_def(svr_resc_def, "mem", 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 = (resi_sum(&pjob->ji_tasks) + 1023) >> 10;	/* in KB */
	*lp = MAX(*lp, lnum);

	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;
	int		fd;
	char		procname[100];
	struct dirent	*dent;
	prpsinfo_t	pi;
	int		sesid;

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