rpp.c

OpenPBS

    int		len;
{
	DOID("write")
	struct	stream	*sp;
	struct	pending	*pp;
	int		hold, residual, more;

	DBPRT((DBTO, "%s: entered index %d size %d\n", id, index, len))
	if (index < 0 || index >= stream_num || len < 0) {
		errno = EINVAL;
		return -1;
	}
	if (len == 0)
		return 0;
	sp = &stream_array[index];
	rpp_stale(sp);			/* check freshness */

	switch (sp->state) {

	case RPP_STALE:
		errno = ETIMEDOUT;
		return -1;

	case RPP_CLOSE_PEND:
		errno = EPIPE;
		return -1;

	case RPP_OPEN_PEND:			/* shouldn't happen */
	case RPP_DEAD:
	case RPP_FREE:
	case RPP_CLOSE_WAIT1:			/* stream closed */
	case RPP_CLOSE_WAIT2:
	case RPP_LAST_ACK:
		errno = ENOTCONN;
		return -1;

	default:
		break;
	}

	residual = 0;
	while (residual < len) {
		hold = sp->pend_attempt % RPP_PKT_DATA;
		if ((pp = sp->pend_tail) == NULL || hold == 0) {
			pp = (struct pending *)malloc(sizeof(struct pending));
			if (sp->pend_tail == NULL)
				sp->pend_head = pp;
			else
				sp->pend_tail->next = pp;
			sp->pend_tail = pp;
			pp->data = (u_char *)malloc(RPP_PKT_SIZE);
			assert(pp->data != NULL);
			pp->next = NULL;
		}
		more = MIN(len - residual, RPP_PKT_DATA - hold);
		memcpy(&pp->data[hold], (char *)buf + residual, more);
		residual += more;
		sp->pend_attempt += more;
	}

	if (rpp_recv_all() == -1)
		return -1;
	rpp_send_out();
	return residual;
}

/*
**	Check a stream to see if it needs attention.
*/
static
int
rpp_attention(index)
    int		index;
{
	DOID("attention")
	int			mesg, count;
	int			seq;
	struct	stream		*sp;
	struct	recv_packet	*pp;

	sp = &stream_array[index];
	DBPRT((DBTO, "%s: stream %d in state %d addr %s\n",
			id, index, sp->state, netaddr(&sp->addr)))
	rpp_stale(sp);

	switch (sp->state) {

	case RPP_STALE:			/* need to report error */
		return TRUE;

	case RPP_CLOSE_PEND:		/* we haven't closed yet */
	case RPP_CONNECT:		/* check for message */
		break;

	default:
		return FALSE;

	}

	if (sp->msg_cnt > 0 && sp->recv_attempt <= sp->msg_cnt)
		return TRUE;		/* message to read */

	mesg = FALSE;
	count = 0;
	for (pp=sp->recv_head, seq=sp->recv_sequence; pp; pp=pp->next, seq++) {
		count += pp->len;
		if (pp->sequence != seq)
			break;
		if (pp->type != RPP_DATA) {	/* end of message */
			mesg = TRUE;
			break;
		}
	}
	if (mesg)
		sp->msg_cnt = count;
	return mesg;
}

/*
**	Check some state before reading or skipping.  If it is
**	okay to continue, return 1.  Otherwise, return <= 0.
*/
static
int
rpp_okay(index)
    int	index;
{
	struct	stream		*sp;
	fd_set			fdset;
	struct	timeval		tv;

	FD_ZERO(&fdset);
	while (rpp_attention(index) == FALSE) {
		int	i;

		tv.tv_sec = RPP_TIMEOUT;
		tv.tv_usec = 0;
		for (i=0; i<rpp_fd_num; i++)
			FD_SET(rpp_fd_array[i], &fdset);
		i = select(FD_SETSIZE, &fdset, NULL, NULL, &tv);
		if (i == -1) 
			return -1;
		if (rpp_recv_all() == -1)
			return -1;
		rpp_send_out();
	}
  
	sp = &stream_array[index];
	if (sp->state == RPP_STALE) {		/* stale output */
		errno = ETIMEDOUT;
		return -1;
	}
	if (sp->recv_attempt == sp->msg_cnt) {	/* end of message */
		if (sp->state == RPP_CLOSE_PEND)
			return -2;
		else
			return 0;
	}
	return 1;
}

/*
**	Read a message.  Return data up to the end of a message
**	or the end of the provided buffer.
**	Return -1 on error, -2 if other side has closed, otherwise
**	number of bytes read.
*/
int
rpp_read(index, buf, len)
    int		index;
    void	*buf;
    int		len;
{
	DOID("read")
	int			hiwater, cpylen, hold, ret, xlen;
	struct	recv_packet	*pp;
	struct	stream		*sp;

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

	if (index < 0 || index >= stream_num || len < 0) {
		errno = EINVAL;
		return -1;
	}
	if (len == 0)
		return 0;
	sp = &stream_array[index];

	switch (sp->state) {

	case RPP_DEAD:
	case RPP_FREE:
	case RPP_CLOSE_WAIT1:
	case RPP_CLOSE_WAIT2:
	case RPP_LAST_ACK:
		errno = ENOTCONN;
		return -1;	/* stream closed */

	default:
		break;
	}

	if ((ret = rpp_okay(index)) <= 0)
		return ret;

	sp = &stream_array[index];
	cpylen = 0;				/* find packet to copy from */
	for (pp=sp->recv_head; pp; pp=pp->next) {
		int	bump = cpylen + pp->len;

		if (sp->recv_attempt < bump)
			break;
		cpylen = bump;
	}
	hiwater = 0;
	xlen = MIN(len, sp->msg_cnt);
	hold = sp->recv_attempt - cpylen;	/* start point in pkt data */
	while (pp && xlen > hiwater) {		/* got room */
		cpylen = MIN(pp->len-hold, xlen-hiwater);
		memcpy((char *)buf + hiwater, &pp->data[hold], cpylen);
		hiwater += cpylen;
		sp->recv_attempt += cpylen;
		hold = 0;
		pp = pp->next;
	}
	return hiwater;
}

/*
**	Commit data which has been read up to recv_attempt if flag
**	is TRUE.  Otherwise, set recv_attempt back to the previous
**	commit point recv_commit.
**	Return -1 on error, FALSE on decommit or if end-of-message has
**	not been reached, TRUE if end-of-message has been reached.
*/
int
rpp_rcommit(index, flag)
    int		index;
    int		flag;
{
	DOID("rcommit")
	struct	stream		*sp;

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

	if (index < 0 || index >= stream_num) {
		errno = EINVAL;
		return -1;
	}
	sp = &stream_array[index];

	switch (sp->state) {

	case RPP_CLOSE_WAIT1:			/* stream closed */
	case RPP_CLOSE_WAIT2:
	case RPP_LAST_ACK:
	case RPP_OPEN_PEND:			/* shouldn't happen */
	case RPP_FREE:
	case RPP_DEAD:
		errno = ENOTCONN;
		return -1;

	default:
		break;
	}

	if (flag == FALSE) {			/* no commit */
		sp->recv_attempt = sp->recv_commit;
		return 0;
	}
	sp->recv_commit = sp->recv_attempt;
	return (sp->recv_commit == sp->msg_cnt);
}

/*
**	Reset end-of-message condition on a stream.  Any packets
**	on the receive queue are freed.
**	Return -1 on error, 0 otherwise.
*/
int
rpp_eom(index)
    int		index;
{
	DOID("eom")
	struct	stream		*sp;
	struct	recv_packet	*pp;

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

	if (index < 0 || index >= stream_num) {
		errno = EINVAL;
		return -1;
	}
	sp = &stream_array[index];
	switch (sp->state) {

	case RPP_CLOSE_WAIT1:			/* stream closed */
	case RPP_CLOSE_WAIT2:
	case RPP_LAST_ACK:
	case RPP_OPEN_PEND:			/* shouldn't happen */
	case RPP_FREE:
	case RPP_DEAD:
		errno = ENOTCONN;
		return -1;

	default:
		break;
	}
/*
**	work though recv packets
*/
	for (pp=sp->recv_head; pp; pp=sp->recv_head) {
		if (pp->type == RPP_GOODBYE)	/* stream finished */
			break;
		if (sp->msg_cnt < pp->len)
			break;
		sp->recv_sequence++;
		sp->msg_cnt -= pp->len;
		if (pp->data)
			free(pp->data);
		sp->recv_head = pp->next;
		free(pp);
	}
	if (sp->recv_head == NULL)
		sp->recv_tail = NULL;
	sp->recv_attempt = 0;
	sp->recv_commit = 0;
	return 0;
}

/*
**	Commit data which has been written up to pend_attempt if flag
**	is TRUE.  Otherwise, set pend_attempt back to the previous
**	commit point pend_commit.
**	Return -1 on error, 0 otherwise.
*/
int
rpp_wcommit(index, flag)
    int		index;
    int		flag;
{
	DOID("wcommit")
	struct	pending		*pp, *next;
	struct	stream		*sp;

	DBPRT((DBTO, "%s: entered index %d\n", id, index))
	if (index < 0 || index >= stream_num) {
		errno = EINVAL;
		return -1;
	}
	sp = &stream_array[index];
	switch (sp->state) {

	case RPP_CLOSE_PEND:
		errno = EPIPE;
		return -1;

	case RPP_STALE:
		errno = ETIMEDOUT;
		return -1;

	case RPP_CLOSE_WAIT1:			/* stream closed */
	case RPP_CLOSE_WAIT2:
	case RPP_LAST_ACK:
	case RPP_OPEN_PEND:			/* shouldn't happen */
	case RPP_FREE:
	case RPP_DEAD:
		errno = ENOTCONN;
		return -1;

	default:
		break;
	}

	if (flag) {			/* commit */
		if (rpp_dopending(index, FALSE))
			return -1;
		if (rpp_recv_all() == -1)
			return -1;
		rpp_send_out();
		return 0;
	}

	sp->pend_attempt = sp->pend_commit;
	if (sp->pend_head == NULL)
		return 0;
	for (pp=sp->pend_head->next; pp; pp=next) {
		free(pp->data);
		next = pp->next;
		free(pp);
	}
	sp->pend_head->next = NULL;
	sp->pend_tail = sp->pend_head;
	return 0;
}

/*
**	Skip len characters of a message.
*/
int
rpp_skip(index, len)
    int		index;
    int		len;
{
	DOID("skip")
	struct	stream		*sp;
	int			ret, hiwater;

	DBPRT((DBTO, "%s: entered index %d\n", id, index))
	if (index < 0 || index >= stream_num) {
		errno = EINVAL;
		return -1;
	}
	sp = &stream_array[index];

	switch (sp->state) {

	case RPP_DEAD:
	case RPP_FREE:
	case RPP_CLOSE_WAIT1:
	case RPP_CLOSE_WAIT2:
	case RPP_LAST_ACK:
		errno = ENOTCONN;
		return -1;	/* stream closed */

	default:
		break;
	}

	if ((ret = rpp_okay(index)) <= 0)
		return ret;

	sp = &stream_array[index];
	hiwater = MIN(sp->msg_cnt - sp->recv_attempt, len);
	sp->recv_attempt += hiwater;
	return hiwater;
}

/*
**	Check for any stream with a message waiting and
**	return the stream number or a -1 if there are none.
*/
int
rpp_poll()
{
	DOID("poll")
	int			i;

	DBPRT((DBTO, "%s: entered streams %d\n", id, stream_num))
	/*
	** Read socket to get any packets
	*/
	for (;;) {
		i = rpp_recv_all();
		if (i == -1 || i == -3)
			break;
	}
	if (i == -1)
		return -1;

	/*
	** See if any stream has a message waiting.
	*/
	for (i=0; i<stream_num; i++) {
		if (rpp_attention(i))
			break;
	}
	if (i < stream_num)	/* found one */
		return i;

	rpp_send_out();
	return -2;
}

/*
**	Process any stream i/o.
**	Return 0 or a -1 if there was an error.
*/
int
rpp_io()
{
	DOID("io")
	int			i;

	DBPRT((DBTO, "%s: entered streams %d\n", id, stream_num))
	/*
	** Read socket to get any packets
	*/
	for (;;) {
		i = rpp_recv_all();
		if (i == -1 || i == -3)
			break;
	}
	if (i == -1)
		return -1;

	rpp_send_out();
	return 0;
}

/*
**	Read a character.
**	Returns  >=0	the char read
**		  -1	error or EOD
**		  -2	EOF
*/
int
rpp_getc(index)
    int		index;
{
	int	ret;
	u_char	c;

	if ((ret = rpp_read(index, &c, 1)) == 1)
		return ((int)c);
	return ((ret == -2) ? -2 : -1);
}

/*
**	Write a character.
*/
int
rpp_putc(index, c)
    int		index;
    int		c;
{
	u_char	x = (u_char)c;

	if (rpp_write(index, &x, 1) != 1)
		return -1;
	return 0;
}