rm.c

openPBS的开放源代码

closerm(stream)
     int	stream;
{
	pbs_errno = 0;
	(void)simplecom(stream, RM_CMD_CLOSE);
	if (delrm(stream) == -1) {
		pbs_errno = ENOTTY;
		return -1;
	}
	return 0;
}

/*
**	Shutdown the resource monitor.  Return result 0 if
**	all is ok or -1 if not (set pbs_errno).
*/
int
downrm(stream)
     int	stream;
{
	pbs_errno = 0;
	if (simplecom(stream, RM_CMD_SHUTDOWN))
		return -1;
	if (simpleget(stream))
		return -1;
	(void)delrm(stream);
	return 0;
}

/*
**	Cause the resource monitor to read the file named.
**	Return the result 0 if all is ok or -1 if not (set pbs_errno).
*/
int
configrm(stream, file)
     int	stream;
     char	*file;
{
	int	ret, len;
	struct	out	*op;

	pbs_errno = 0;
	if ((op = findout(stream)) == NULL)
		return -1;
	op->len = -1;

	if (file[0] != '/' || (len = strlen(file)) > (size_t)MAXPATHLEN) {
		pbs_errno = EINVAL;
		return -1;
	}

	if (startcom(stream, RM_CMD_CONFIG) != DIS_SUCCESS)
		return -1;
	ret = diswcs(stream, file, len);
	if (ret != DIS_SUCCESS) {
#if	defined(ECOMM)
		pbs_errno = ECOMM;
#elif	defined(ENOCONNECT)
		pbs_errno = ENOCONNECT;
#else
		pbs_errno = ETXTBSY;
#endif
		DBPRT(("configrm: diswcs %s\n", dis_emsg[ret]))
		return -1;
	}
	if (flush_dis(stream) == -1) {
		pbs_errno = errno;
		DBPRT(("configrm: flush error %d\n", pbs_errno))
		return -1;
	}

	if (simpleget(stream))
		return -1;
	return 0;
}

/*
**	Begin a new message to the resource monitor if necessary.
**	Add a line to the body of an outstanding command to the resource
**	monitor.
**	Return the result 0 if all is ok or -1 if not (set pbs_errno).
*/
static
int
doreq(op, line)
     struct	out	*op;
     char	*line;
{
	int	ret;

	if (op->len == -1) {	/* start new message */
		if (startcom(op->stream, RM_CMD_REQUEST) != DIS_SUCCESS)
			return -1;
		op->len = 1;
	}
	ret = diswcs(op->stream, line, strlen(line));
	if (ret != DIS_SUCCESS) {
#if	defined(ECOMM)
		pbs_errno = ECOMM;
#elif	defined(ENOCONNECT)
		pbs_errno = ENOCONNECT;
#else
		pbs_errno = ETXTBSY;
#endif
		DBPRT(("doreq: diswcs %s\n", dis_emsg[ret]))
		return -1;
	}
	return 0;
}

/*
**	Add a request to a single stream.
*/
int
addreq(stream, line)
     int	stream;
     char	*line;
{
	struct	out	*op;

	pbs_errno = 0;
	if ((op = findout(stream)) == NULL)
		return -1;
	funcs_dis();
	if (doreq(op, line) == -1) {
		(void)delrm(stream);
		return -1;
	}
	return 0;
}

/*
**	Add a request to every stream.
**	Return the number of streams acted upon.
*/
int
allreq(line)
     char	*line;
{
	struct	out	*op, *prev;
	int		i, num;

	funcs_dis();
	pbs_errno = 0;
	num = 0;
	for (i=0; i<HASHOUT; i++) {
		prev=NULL;
		op=outs[i];
		while (op) {
			if (doreq(op, line) == -1) {
				struct	out	*hold = op;

				close_dis(op->stream);
				if (prev)
					prev->next = op->next;
				else
					outs[i] = op->next;
				op = op->next;
				free(hold);
			}
			else {
				prev = op;
				op = op->next;
				num++;
			}
		}
	}
	return num;
}

/*
**	Finish (and send) any outstanding message to the resource monitor.
**	Return a pointer to the next response line or a NULL if
**	there are no more or an error occured.  Set pbs_errno on error.
*/
char *
getreq(stream)
     int	stream;
{
	char	*startline;
	struct	out	*op;
	int	ret;

	pbs_errno = 0;
	if ((op = findout(stream)) == NULL)
		return NULL;
	if (op->len >= 0) {	/* there is a message to send */
		if (flush_dis(stream) == -1) {
			pbs_errno = errno;
			DBPRT(("getreq: flush error %d\n", pbs_errno))
			(void)delrm(stream);
			return NULL;
		}
		op->len = -2;
#if	RPP
		(void)rpp_eom(stream);
#endif
	}
	funcs_dis();
	if (op->len == -2) {
		if (simpleget(stream) == -1)
			return NULL;
		op->len = -1;
	}
	startline = disrst(stream, &ret);
	if (ret == DIS_EOF) {
		return NULL;
	}
	else if (ret != DIS_SUCCESS) {
		pbs_errno = errno ? errno : EIO;
		DBPRT(("getreq: cannot read string %s\n", dis_emsg[ret]))
		return NULL;
	}

	if (!full) {
		char	*cc, *hold;
		int	indent = 0;

		for (cc=startline; *cc; cc++) {
			if (*cc == '[')
				indent++;
			else if (*cc == ']')
				indent--;
			else if (*cc == '=' && indent == 0) {
				hold = strdup(cc + 1);
				free(startline);
				startline = hold;
				break;
			}
		}
	}
	return startline;
}

/*
**	Finish and send any outstanding messages to all resource monitors.
**	Return the number of messages flushed.
*/
int
flushreq()
{
	struct	out	*op, *prev;
	int	did, i;

	pbs_errno = 0;
	did = 0;
	for (i=0; i<HASHOUT; i++) {
		for (op=outs[i]; op; op=op->next) {
			if (op->len <= 0)	/* no message to send */
				continue;
			if (flush_dis(op->stream) == -1) {
				pbs_errno = errno;
				DBPRT(("flushreq: flush error %d\n", pbs_errno))
				close_dis(op->stream);
				op->stream = -1;
				continue;
			}
			op->len = -2;
#if	RPP
			(void)rpp_eom(op->stream);
#endif
			did++;
		}

		prev = NULL;
		op = outs[i];
		while (op) {		/* get rid of bad streams */
			if (op->stream != -1) {
				prev = op;
				op = op->next;
				continue;
			}
			if (prev == NULL) {
				outs[i] = op->next;
				free(op);
				op = outs[i];
			}
			else {
				prev->next = op->next;
				free(op);
				op = prev->next;
			}
		}
	}
	return did;
}

/*
**	Return the stream number of the next stream with something
**	to read or a negative number (the return from rpp_poll)
**	if there is no stream to read.
*/
int
activereq()
{
	static	char	id[] = "activereq";
	struct	out	*op;
	int		try, i, num;
	int		bucket;
	struct	timeval	tv;
	fd_set		fdset;

	pbs_errno = 0;
	flushreq();
	FD_ZERO(&fdset);

#if	RPP
	for (try=0; try<3;) {
		if ((i = rpp_poll()) >= 0) {
			if ((op = findout(i)) != NULL)
				return i;

			op = (struct out *)malloc(sizeof(struct out));
			if (op == NULL) {
				pbs_errno = errno;
				return -1;
			}

			bucket = i % HASHOUT;
			op->stream = i;
			op->len = -2;
			op->next = outs[bucket];
			outs[bucket] = op;
		}
		else if (i == -1) {
			pbs_errno = errno;
			return -1;
		}
		else {
			extern	int	rpp_fd;

			FD_SET(rpp_fd, &fdset);
			tv.tv_sec = 5;
			tv.tv_usec = 0;
			num = select(FD_SETSIZE, &fdset, NULL, NULL, &tv);
			if (num == -1) {
				pbs_errno = errno;
				DBPRT(("%s: select %d\n", id, pbs_errno))
				return -1;
			}
			if (num == 0) {
				try++;
				DBPRT(("%s: timeout %d\n", id, try))
			}
		}
	}
	return i;
#else
	pbs_errno = 0;
	for (i=0; i<HASHOUT; i++) {
		struct	out	*op;

		op=outs[i];
		while (op) {
			FD_SET(op->stream, &fdset);
			op = op->next;
		}
	}
	tv.tv_sec = 15;
	tv.tv_usec = 0;
	num = select(FD_SETSIZE, &fdset, NULL, NULL, &tv);
	if (num == -1) {
		pbs_errno = errno;
		DBPRT(("%s: select %d\n", id, pbs_errno))
		return -1;
	}
	else if (num == 0)
		return -2;

	for (i=0; i<HASHOUT; i++) {
		struct	out	*op;

		op=outs[i];
		while (op) {
			if (FD_ISSET(op->stream, &fdset))
				return op->stream;
			op = op->next;
		}
	}
	return -2;
#endif
}

/*
**	If flag is true, turn on "full response" mode where getreq
**	returns a pointer to the beginning of a line of response.
**	This makes it possible to examine the entire line rather
**	than just the answer following the equal sign.
*/
void
fullresp(flag)
     int	flag;
{
#if	RPP
	extern	int	rpp_dbprt;

	if (flag)
		rpp_dbprt = 1 - rpp_dbprt;	/* toggle RPP debug */
#endif
	pbs_errno = 0;
	full = flag;
	return;
}