/*
 * Copyright (C) 1994-2018 Altair Engineering, Inc.
 * For more information, contact Altair at www.altair.com.
 *
 * This file is part of the PBS Professional ("PBS Pro") software.
 *
 * Open Source License Information:
 *
 * PBS Pro is free software. You can redistribute it and/or modify it under the
 * terms of the GNU Affero General Public License as published by the Free
 * Software Foundation, either version 3 of the License, or (at your option) any
 * later version.
 *
 * PBS Pro is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
 * FOR A PARTICULAR PURPOSE.
 * See the GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 * Commercial License Information:
 *
 * For a copy of the commercial license terms and conditions,
 * go to: (http://www.pbspro.com/UserArea/agreement.html)
 * or contact the Altair Legal Department.
 *
 * Altair’s dual-license business model allows companies, individuals, and
 * organizations to create proprietary derivative works of PBS Pro and
 * distribute them - whether embedded or bundled with other software -
 * under a commercial license agreement.
 *
 * Use of Altair’s trademarks, including but not limited to "PBS™",
 * "PBS Professional®", and "PBS Pro™" and Altair’s logos is subject to Altair's
 * trademark licensing policies.
 *
 */
/**
 * @file	rpp.c
 * @brief
 *	Routines to communicate with UDP packets - Reliable Packet Protocol.
 *
 *	This package provides for sending information in "messages"
 *	which are complete blocks of data which will either arrive
 *	complete or not at all.
 */

#include <pbs_config.h>   /* the master config generated by configure */

#include	<stdio.h>
#include	<stddef.h>
#include	<stdlib.h>
#include	<string.h>
#include	<unistd.h>
#include	<fcntl.h>
#include	<netdb.h>
#include	<errno.h>
#include	<assert.h>
#include	<math.h>
#include	<sys/types.h>
#include	<sys/file.h>
#include	<sys/stat.h>
#include	<sys/socket.h>
#include	<sys/time.h>
#include	<time.h>
#include	<netinet/in.h>
#include	"rpp.h"
#include	"pbs_internal.h"
#include	"pbs_error.h"
#include	"dis_init.h"

#if !defined(H_ERRNO_DECLARED)
extern int h_errno;
#endif

/* definitions of pointer functions for global use */
int                     (*pfn_rpp_open)(char *, unsigned int);
int                     (*pfn_rpp_bind)(unsigned int);
int                     (*pfn_rpp_poll)(void);
int                     (*pfn_rpp_io)(void);
int                     (*pfn_rpp_read)(int, void *, int);
int                     (*pfn_rpp_write)(int, void *, int);
int                     (*pfn_rpp_close)(int);
void					(*pfn_rpp_destroy)(int);
struct   sockaddr_in*   (*pfn_rpp_localaddr)(int);
struct   sockaddr_in*   (*pfn_rpp_getaddr)(int);
int                     (*pfn_rpp_flush)(int);
void                    (*pfn_rpp_shutdown)(void);
void                    (*pfn_rpp_terminate)(void);
int                     (*pfn_rpp_rcommit)(int, int);
int                     (*pfn_rpp_wcommit)(int, int);
int                     (*pfn_rpp_skip)(int, size_t);
int                     (*pfn_rpp_eom)(int);
int                     (*pfn_rpp_getc)(int);
int                     (*pfn_rpp_putc)(int, int);
void			 		(*pfn_DIS_rpp_reset)();
void			 		(*pfn_DIS_rpp_setup)(int);
void 					(*pfn_rpp_add_close_func)(int, void (*func)(int));

/* forward declarations */
void 			__DIS_rpp_setup(int);
void 			__DIS_rpp_reset(void);
int                     __rpp_open(char *, unsigned int);
int                     __rpp_bind(unsigned int);
int                     __rpp_poll(void);
int                     __rpp_io(void);
int                     __rpp_read(int, void *, int);
int                     __rpp_write(int, void *, int);
int                     __rpp_close(int);
struct   sockaddr_in*   __rpp_localaddr(int);
struct   sockaddr_in*   __rpp_getaddr(int);
int                     __rpp_flush(int);
void                    __rpp_shutdown(void);
void                    __rpp_terminate(void);
int                     __rpp_rcommit(int, int);
int                     __rpp_wcommit(int, int);
int                     __rpp_skip(int, size_t);
int                     __rpp_eom(int);
int                     __rpp_getc(int);
int                     __rpp_putc(int, int);
void			__DIS_rpp_reset();
void			__DIS_rpp_setup(int);

/**
 * @brief
 *	-wrapper function for __DIS_rpp_reset.
 *
 * @param[in] stream - socket stream
 *
 */
void
__DIS_rpp_setup(int stream)
{
	__DIS_rpp_reset();
}

/*
 *	Boolean Constants
 */
#ifndef	FALSE
#define	FALSE	0
#endif
#ifndef	TRUE
#define	TRUE	1
#endif

/*
 *	Turn on for debug log dump to pbs_tmpdir.
 #define	RPPLOG	1
 */

/*
 *	Size-Constants for the various parts of RPP packet
 */
#ifdef NAS_CLUSTER /* localmod 087 */
#define	RPP_PKT_SIZE	32*1024 /* max packet length, including header */
#else
#define	RPP_PKT_SIZE	4*1024 /* max packet length, including header */
#endif /* localmod 087 */
#define	RPP_PKT_HEAD	26     /* Size of packet header, includes CRC */
#define	RPP_HDR_SID	2      /* position of stream id in packet header */
#define	RPP_HDR_SEQ	10     /* position of "sequence" in pkt header */
#define	RPP_CRC_LEN	8      /* Length of CRC field in pkt header */

#define RPP_PKT_CRC	(RPP_PKT_HEAD - RPP_CRC_LEN)
#define	RPP_PKT_DATA	(RPP_PKT_SIZE - RPP_PKT_HEAD)


/*
 *	RPP packet-header fields
 *
 *   length             field
 *  ========          =========
 *     2		type
 *     8		stream id
 *     8		sequence
 *     8		crc
 */


/*
 *	Integer codes for all the valid RPP message types
 */
#define	RPP_ACK		1
#define	RPP_DATA	2
#define	RPP_EOD		3
#define	RPP_HELLO1	4
#define	RPP_HELLO2	5
#define	RPP_GOODBYE	6


/*
 *	Integer codes for all the valid RPP state values
 */
#define RPP_DEAD	-1
#define	RPP_FREE 	 0
#define	RPP_OPEN_PEND	 1
#define	RPP_OPEN_WAIT	 2
#define	RPP_CONNECT	 3
#define	RPP_CLOSE_PEND	 4
#define	RPP_LAST_ACK	 5
#define	RPP_CLOSE_WAIT1	 6
#define	RPP_CLOSE_WAIT2	 7
#define	RPP_STALE	99

/**
 *	Time in seconds; packet on the master send queue is not sent more
 *	often than every RPP_TIMEOUT seconds.
 */
#define	RPP_TIMEOUT	2

/*
 *	maximum and minimum values for SO_RCVBUF size in bytes.
 */
#define	RPP_MAX_SO_RCVBUF	2097152	/* 2MB */
#define RPP_MIN_SO_RCVBUF	131071  /* 128KB - 1 */

/**
 *	Several kinds of linked lists hang from each RPP stream structure.
 *	In particular, each stram structure has a list of send_packet
 *	structures.  These structures record information that's necessary for
 *	managing a piece of data that's being sent to the other end of the
 *	connection.  Besides having a pointer that links it to the next
 *	send_packet on the stream, a send_packet posseses a pair of pointers
 *	(up,down) which link the  send_packet to the master send list when
 *	connection sequencing has reached the point where it is now proper to
 *	attach it to the list (stream is fully opened).  Once on the master
 *	sendlist, any invocation of the __rpp_send_out routine will attempt to
 *	transfer the send_packet's data buffer to the other end of the stream
 *	connection if all of the following hold:
 *
 *	1) packet hasn't reached its maximum-transfer-attempt limit
 *	2) RPP_TIMEOUT or more seconds have elapsed since the last attempt
 *	3) the transfer has yet to be ACK'd by the other side.
 *	4) less than rpp_highwater number of non-duplicate packets are
 *	   un-ACK'd
 *
 *	Those "send_packets" that deal with stream control
 *	(RPP_ACK, RPP_HELLO1, RPP_HELLO2, RPP_GOODBYE) have no
 *	associated data, the send_packet's data buffer is comprised
 *	of only the header.
 */

struct	send_packet {
	u_char	*data;		/* points to a buffer to be transferred */

	u_short	type;		/* RPP_ACK, RPP_HELLO1, RPP_DATA, etc */

	u_short	retries;	/* number of send retries; <= rpp_retry */

	int	len;		/* size *data not counting the header */

	int	index;		/* other end's stream id */

	int	sequence;	/* sequential value that is placed into */
	/* data buffer's header and is returned */
	/* in the RPP_ACK; identifies the data */
	/* that was transferred and now needs */
	/* to be removed from master send list */

	time_t	time_sent;	/* time packet was last sent; zero if */
	/* it has yet to be sent */
	time_t	first_sent;

	struct	send_packet	*next;	/* next packet on stream's send list */

	struct	send_packet	*up;	/* used when the send_packet is */
	/* on RPP's master send list */
	/* pointer to preceeding packet */

	struct	send_packet	*down;	/* similar to up, but pointing to */
	/* packet following this one */
};

/**
 *	Hanging from an stream structure is a linked list of recv_packets
 *	Each recv_packet records the information necessary for managing a
 *	piece of data that was received from the other end of a connection.
 *	With the exception of RPP_GOODBYE (a surrogate RPP_EOD), pkts that deal
 *	with stream control (RPP_ACK,RPP_HELLO1,RPP_HELLO2,) don't ever make it
 *	to the stream's receive list, they are handled directly as they come
 *	in.  So, the only types of packets on a stream's receive list should
 *	be ones of type RPP_DATA, RPP_EOD, and RPP_GOODBYE.
 */

struct	recv_packet {
	u_char	*data;		/* pointer to byte string that was sent from */
	/* the other end of the stream */

	u_short	type;		/* RPP_DATA, RPP_EOD or RPP_GOODBYE; i.e. a */
	/* piece of the message or the end of the */
	/* message (RPP_EOD or RPP_GOODBYE) */

	int	len;		/* size in bytes of the received string; */
	/* does not include the length of the header */

	int	sequence;	/* decoded sequential number; it describes */
	/* the buffer's sequential output order */
	/* relative to the other buffers */

	struct	recv_packet  *next;	/* pointer to next recv_packet in */
	/* the linked  receive list for */
	/* this stream */
};


/**
 *	Each stream structure has a linked list of pend structs hanging
 *	from it.  This list of structs is used to manage a set of data buffers
 *	that may or may not all get 'packetized' and sent over to the other
 *	end of the connection.  Subsequent to the creation of the list of data
 *	buffers, the creating program decides to either commit or not commit
 *	the sending of this data to the other side-- done by calling the
 *	interface function, __rpp_wcommit.  The interface function
 *	__rpp_write doesn't send the data, it merely
 *	attaches the data buffer to a pending struct and links this struct to
 *	the stream's list of pending structs.  No transfer to the other end
 *	gets set in motion by calling __rpp_write.
 *
 *	Doing an __rpp_wcommit, on the other hand, entails adjoining a header
 *	to each pending struct's data buffer, attaching the buffer to a
 *	new send_packet struct, linking the send_packet to the master send
 *	list, freeing the pend struct, and doing an update of the stream's
 *	pend_commit variable -- a running sum of the number of bytes sent
 *	to the other side.
 *
 *	Decommiting data that was written for transfer to the other end entails
 *	removing and freeing the pending structs and their associated data
 *	buffer from the stream's pend list and, updating the stream's
 *	pend_attempt varable back to the byte count that is stored in its
 *	pend_commit variable.  Refer to relevant fields in struct stream.
 */

struct	pending {
	u_char	*data;		/* pointer to a buffer of user data */
	struct	pending	*next;	/* pointer to the next pending struct */
};


/**
 *	Every stream that comes into existence during the life of the
 *	process gets realized on each end of the connection by a stream
 *	struct.  All the stream structs generated by a process are in a
 *	dynamic array in the process' heap area.
 *	Each stream is a finite state machine.
 */

struct	stream {
	int			state;		/* state of this end of the */
	/* connection; RPP_OPEN, etc */

	struct	sockaddr_in	addr;		/* address of the other end; */
	/* port/family/IPadrs */

	struct	in_addr		*addr_array;	/* array of alternate network */
	/* addresses for other end */
	/* of the connection */

	int			fd;		/* must be in rpp_fd_array */

	int			stream_id;	/* id of other end of the */
	/* connection; array position */
	/* of stream struct on the */
	/* other end */

	int			open_key;	/* unique bit pattern created */
	/* by the end issuing the */
	/* __rpp_open.  It's the same */
	/* for each end of the */
	/* connecton; used in setting */
	/* up the stream connection */

	int			msg_cnt;	/* size in bytes of current */
	/* DATA/EOD/GOODBYE message */

	int			send_sequence;	/* initialized to value of 1 */
	/* and incremented by 1 for */
	/* each packet that's added */
	/* to the master send list */

	struct	pending		*pend_head;	/* head and tail pointers for */
	struct	pending		*pend_tail;	/* stream's pend list; see */
	/* struct pend definition */

	int			pend_commit;	/* total number of data bytes */
	/* sent to other end */
	int			pend_attempt;	/* total number bytes that */
	/* reside in the list of */
	/* pending struct buffers. */
	/* pend_commit<=pend_attempt */

	struct	send_packet	*send_head;	/* head and tail pointers for */
	struct	send_packet	*send_tail;	/* stream's master send list */
	/* see struct send_packet */

	int			recv_sequence;	/* monotonic,increasing, by 1 */
	/* starts from zero; A packet */
	/* on the stream's recv list */
	/* having a sequence number */
	/* less than this value is a */
	/* packet of an earlier mesg */

	struct	recv_packet	*recv_head;	/* head and tail pointers for */
	struct	recv_packet	*recv_tail;	/* the stream's recv list; */
	/* see struct recv_packet */

	int			recv_commit;	/* number of bytes from */
	/* start of current message */
	/* that have been accepted */
	/* by the reader on this end */

	int			recv_attempt;	/* number bytes, from start */
	/* of current message, that */
	/* have been read */
};

/*
 *	Static Variables
 */
static	struct	stream	*stream_array = NULL;	/* pointer to stream struct */
/* dynamic array */

static	int		stream_num = 0;		/* current number of stream */
/* structs in stream_array */

static	int		pkts_sent = 0;		/* range: 0 - rpp_highwater; */
/* incremented with each new */
/* pkt sent; retransmissions */
/* are not counted */

static	int		open_key = 0;		/* monotonicly increasing */
/* value stored in stream */
/* struct on __rpp_open and */
/* passed to other end to */
/* be recorded in */
/* corresponding stream */
/* struct there */

static	struct	send_packet	*top = NULL;	/* ptrs to beginning and end */
static	struct	send_packet	*bottom = NULL; /* of master send list; */
/* All sent data is linked */
/* between top and bottom. */

/*
 *	Track statistics
 */
#define	RPP_LOG_PERIOD	3600	/* one hour */

static	u_long	total_fails = 0;
static	u_long	total_retries = 0;
static	u_long	total_packets = 0;
static	u_long	total_errors = 0;

static	u_long	last_fails = 0;
static	u_long	last_retries = 0;
static	u_long	last_packets = 0;
static	u_long	last_errors = 0;

static	time_t	last_time = 0;

static	char	rpp_logbuf[512];

static	int	rpp_cur_rcvbuf_size = -1;
static	int	rpp_max_rcvbuf_size = RPP_MAX_SO_RCVBUF;
static	int	rpp_min_rcvbuf_size = RPP_MIN_SO_RCVBUF;

/*
 *	Global Variables
 */
pid_t	bind_pid = (pid_t)-1;			/* pid who did __rpp_bind */
int	rpp_dbprt = 0;				/* controls debug printing */

rpp_logfunc_t	*rpp_logfunc = NULL;		/* log function */

/**
 *	Current file descriptor.  Any call to __rpp_open will use this
 *	for the returned stream.
 */
int		rpp_fd = -1;

/**
 *	A dynamic array of socket descriptors bound to a network address.
 *	More than one call to __rpp_bind can result in this having multiple
 *	entries.  The value of rpp_fd will be contained in this array.
 */
int		*rpp_fd_array = NULL;

/**
 *	Number of elements in rpp_fd_array
 */
int		rpp_fd_num = 0;


/**
 *	Number of retrys to for each packet.
 */
int		rpp_retry = RPP_RETRY;

/**
 *	Number of packets to send before gettin an ACK.
 */
int		rpp_highwater = RPP_HIGHWATER;

/**
 * @brief
 *	Read timeout in seconds
 *
 * Initialized to -1 which means "don't do a timeout".  Positive values
 * will cause __rpp_read to return an error after the given number of
 * seconds have passed without reading any data.
 *
 * @see
 *	rpp_advise
 * @see
 *	__rpp_okay
 * @see
 *	__rpp_read
 */
static	time_t	read_timeout = -1;

#define	XX	(char)(-1)

/**
 *	Tables used by the macros I2TOH, HTOI2, I8TOH, HTOI8
 *	to convert 2 and 8 digit hexidecimal strings to integer and back.
 */

char	cval[] = {
	XX, XX, XX, XX, XX, XX, XX, XX,		/* nul .. bel */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* bs .. si */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* dle .. etb */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* can .. us */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* sp .. ' */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* ( .. / */
	0,  1,  2,  3,  4,  5,  6,  7,		/* 0 .. 7 */
	8,  9, XX, XX, XX, XX, XX, XX,		/* 8 .. ? */
	XX, 10, 11, 12, 13, 14, 15, XX,		/* @ .. G */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* H .. O */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* P .. W */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* X .. _ */
	XX, 10, 11, 12, 13, 14, 15, XX,		/* ` .. g */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* h .. o */
	XX, XX, XX, XX, XX, XX, XX, XX,		/* p .. w */
	XX, XX, XX, XX, XX, XX, XX, XX		/* x .. del */
};

char	ival[] = {
	'0', '1', '2', '3', '4', '5', '6', '7',
	'8', '9', 'A', 'B', 'C', 'D', 'E', 'F'
};

/*
 *	Conversion macros
 */

#define	I2TOH(i, h) \
{ \
	int	num = i; \
	char	*str = h; \
	str[1] = ival[num & 0xF]; num >>= 4 ; \
	str[0] = ival[num & 0xF]; \
}

#define	I8TOH(i, h) \
{ \
	u_long	num = i; \
	char	*str = h; \
	str[7] = ival[num & 0xF]; num >>= 4; \
	str[6] = ival[num & 0xF]; num >>= 4; \
	str[5] = ival[num & 0xF]; num >>= 4; \
	str[4] = ival[num & 0xF]; num >>= 4; \
	str[3] = ival[num & 0xF]; num >>= 4; \
	str[2] = ival[num & 0xF]; num >>= 4; \
	str[1] = ival[num & 0xF]; num >>= 4; \
	str[0] = ival[num & 0xF]; \
}

#define	HTOI2(h, i) \
{ \
	char	*str = h; \
	int	num = 0; \
	num = cval[str[0] & 0xFF]; num <<= 4; \
	num |= cval[str[1] & 0xFF]; \
	i = num; \
}

#define	HTOI8(h, i) \
{ \
	char	*str = h; \
	u_long	num; \
	num  = (long)cval[str[0] & 0x7F]; num <<= 4; \
	num |= (long)cval[str[1] & 0x7F]; num <<= 4; \
	num |= (long)cval[str[2] & 0x7F]; num <<= 4; \
	num |= (long)cval[str[3] & 0x7F]; num <<= 4; \
	num |= (long)cval[str[4] & 0x7F]; num <<= 4; \
	num |= (long)cval[str[5] & 0x7F]; num <<= 4; \
	num |= (long)cval[str[6] & 0x7F]; num <<= 4; \
	num |= (long)cval[str[7] & 0x7F]; \
	i = num; \
}


/*
 *	Different print macros for use in debugging.
 */

#ifdef  DEBUG

#define	DBTO	stdout
#define	DBPRT(x) \
	if (rpp_dbprt) { \
		int	err = errno; \
		fprintf(DBTO, "%lX: ", time(0)); \
       		fprintf x; \
		errno = err; \
	}

#elif	defined(RPPLOG)

static	char	*blog_buf = NULL;
static	int	blog_buflen = 0;

static	int	blog_head = 0;

/**
 * @brief
 *	-initialize the log buffer with input string
 *
 * @param[in] s - log string
 * @param[in] len - length of string
 *
 * @return 	Void
 *
 */
void
blog_init(char *s, int len)
{
	if (blog_buf != NULL || len <= 0)
		return;

	if (s == NULL)
		blog_buf = (char *)malloc(len);
	else
		blog_buf = s;
	blog_buflen = len;
}

/**
 * @brief
 *	-write buffer log with input string
 *
 * @param[in] s - string
 *
 * @return	int
 * @retval	length of buf	success
 * @retval	0		error
 *
 */
int
blog_write(char *s)
{
	int	i, len;

	if (s == NULL || *s == '\0')
		return 0;

	if (blog_buf == NULL)
		blog_init(NULL, 64*1024);

	len = strlen(s) + 1;
	if (len > blog_buflen)
		return -1;

	for (i=0; i<len; i++) {
		blog_buf[blog_head++] = *s++;
		blog_head %= blog_buflen;
	}
	return len-1;
}

/**
 * @brief
 *	-output the content of log buffer into file.
 *
 * @param[in] filename - filename where log output written
 *
 * @return	Void
 *
 */
void
blog_out(char *filename)
{
	FILE	*f;
	int	btrail;
	int	c;

	if (blog_buf == NULL)
		return;

	if ((f = fopen(filename, "a")) == NULL)
		return;

#ifdef WIN32
	secure_file(filename, "Administrators",
		READS_MASK|WRITES_MASK|STANDARD_RIGHTS_REQUIRED);
#endif

	for (btrail = blog_head; blog_buf[btrail] != '\0'; btrail++)
		btrail %= blog_buflen;

	for (btrail++; btrail != blog_head; btrail++) {
		btrail %= blog_buflen;

		c = (int)blog_buf[btrail];
		if (c == '\0')
			continue;

		putc(c, f);
	}
	fclose(f);
	return;
}

static	char	logbuf[4096], *logp;

#define DBTO	logp

#define	DBPRT(x) \
	sprintf(logbuf, "%lX: ", time(0)); \
	logp = logbuf + strlen(logbuf); \
	sprintf x; \
	blog_write(logbuf); \
	if (rpp_dbprt) { \
		int	err = errno; \
		sprintf(logbuf, "%s/rpp_log.%d", pbs_conf.pbs_tmpdir, getpid()); \
		blog_out(logbuf); \
       		rpp_dbprt = 0; \
		errno = err; \
	}

#else
#define	DBTO	xxx
#define	DBPRT(x)
#endif


#ifndef	MIN
#define	MIN(x, y)	(((x) < (y)) ? (x) : (y))
#endif
#ifndef	MAX
#define	MAX(x, y)	(((x) > (y)) ? (x) : (y))
#endif


/*
 *	BEGIN included source
 */
/*-
 * Copyright (c) 1991, 1993
 *	The Regents of the University of California.  All rights reserved.
 *
 * This code is derived from software contributed to Berkeley by
 * James W. Williams of NASA Goddard Space Flight Center.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. All advertising materials mentioning features or use of this software
 *    must display the following acknowledgment:
 *	This product includes software developed by the University of
 *	California, Berkeley and its contributors.
 * 4. Neither the name of the University nor the names of its contributors
 *    may be used to endorse or promote products derived from this software
 *    without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */

static u_long crctab[] = {
	0x0,
	0x04c11db7, 0x09823b6e, 0x0d4326d9, 0x130476dc, 0x17c56b6b,
	0x1a864db2, 0x1e475005, 0x2608edb8, 0x22c9f00f, 0x2f8ad6d6,
	0x2b4bcb61, 0x350c9b64, 0x31cd86d3, 0x3c8ea00a, 0x384fbdbd,
	0x4c11db70, 0x48d0c6c7, 0x4593e01e, 0x4152fda9, 0x5f15adac,
	0x5bd4b01b, 0x569796c2, 0x52568b75, 0x6a1936c8, 0x6ed82b7f,
	0x639b0da6, 0x675a1011, 0x791d4014, 0x7ddc5da3, 0x709f7b7a,
	0x745e66cd, 0x9823b6e0, 0x9ce2ab57, 0x91a18d8e, 0x95609039,
	0x8b27c03c, 0x8fe6dd8b, 0x82a5fb52, 0x8664e6e5, 0xbe2b5b58,
	0xbaea46ef, 0xb7a96036, 0xb3687d81, 0xad2f2d84, 0xa9ee3033,
	0xa4ad16ea, 0xa06c0b5d, 0xd4326d90, 0xd0f37027, 0xddb056fe,
	0xd9714b49, 0xc7361b4c, 0xc3f706fb, 0xceb42022, 0xca753d95,
	0xf23a8028, 0xf6fb9d9f, 0xfbb8bb46, 0xff79a6f1, 0xe13ef6f4,
	0xe5ffeb43, 0xe8bccd9a, 0xec7dd02d, 0x34867077, 0x30476dc0,
	0x3d044b19, 0x39c556ae, 0x278206ab, 0x23431b1c, 0x2e003dc5,
	0x2ac12072, 0x128e9dcf, 0x164f8078, 0x1b0ca6a1, 0x1fcdbb16,
	0x018aeb13, 0x054bf6a4, 0x0808d07d, 0x0cc9cdca, 0x7897ab07,
	0x7c56b6b0, 0x71159069, 0x75d48dde, 0x6b93dddb, 0x6f52c06c,
	0x6211e6b5, 0x66d0fb02, 0x5e9f46bf, 0x5a5e5b08, 0x571d7dd1,
	0x53dc6066, 0x4d9b3063, 0x495a2dd4, 0x44190b0d, 0x40d816ba,
	0xaca5c697, 0xa864db20, 0xa527fdf9, 0xa1e6e04e, 0xbfa1b04b,
	0xbb60adfc, 0xb6238b25, 0xb2e29692, 0x8aad2b2f, 0x8e6c3698,
	0x832f1041, 0x87ee0df6, 0x99a95df3, 0x9d684044, 0x902b669d,
	0x94ea7b2a, 0xe0b41de7, 0xe4750050, 0xe9362689, 0xedf73b3e,
	0xf3b06b3b, 0xf771768c, 0xfa325055, 0xfef34de2, 0xc6bcf05f,
	0xc27dede8, 0xcf3ecb31, 0xcbffd686, 0xd5b88683, 0xd1799b34,
	0xdc3abded, 0xd8fba05a, 0x690ce0ee, 0x6dcdfd59, 0x608edb80,
	0x644fc637, 0x7a089632, 0x7ec98b85, 0x738aad5c, 0x774bb0eb,
	0x4f040d56, 0x4bc510e1, 0x46863638, 0x42472b8f, 0x5c007b8a,
	0x58c1663d, 0x558240e4, 0x51435d53, 0x251d3b9e, 0x21dc2629,
	0x2c9f00f0, 0x285e1d47, 0x36194d42, 0x32d850f5, 0x3f9b762c,
	0x3b5a6b9b, 0x0315d626, 0x07d4cb91, 0x0a97ed48, 0x0e56f0ff,
	0x1011a0fa, 0x14d0bd4d, 0x19939b94, 0x1d528623, 0xf12f560e,
	0xf5ee4bb9, 0xf8ad6d60, 0xfc6c70d7, 0xe22b20d2, 0xe6ea3d65,
	0xeba91bbc, 0xef68060b, 0xd727bbb6, 0xd3e6a601, 0xdea580d8,
	0xda649d6f, 0xc423cd6a, 0xc0e2d0dd, 0xcda1f604, 0xc960ebb3,
	0xbd3e8d7e, 0xb9ff90c9, 0xb4bcb610, 0xb07daba7, 0xae3afba2,
	0xaafbe615, 0xa7b8c0cc, 0xa379dd7b, 0x9b3660c6, 0x9ff77d71,
	0x92b45ba8, 0x9675461f, 0x8832161a, 0x8cf30bad, 0x81b02d74,
	0x857130c3, 0x5d8a9099, 0x594b8d2e, 0x5408abf7, 0x50c9b640,
	0x4e8ee645, 0x4a4ffbf2, 0x470cdd2b, 0x43cdc09c, 0x7b827d21,
	0x7f436096, 0x7200464f, 0x76c15bf8, 0x68860bfd, 0x6c47164a,
	0x61043093, 0x65c52d24, 0x119b4be9, 0x155a565e, 0x18197087,
	0x1cd86d30, 0x029f3d35, 0x065e2082, 0x0b1d065b, 0x0fdc1bec,
	0x3793a651, 0x3352bbe6, 0x3e119d3f, 0x3ad08088, 0x2497d08d,
	0x2056cd3a, 0x2d15ebe3, 0x29d4f654, 0xc5a92679, 0xc1683bce,
	0xcc2b1d17, 0xc8ea00a0, 0xd6ad50a5, 0xd26c4d12, 0xdf2f6bcb,
	0xdbee767c, 0xe3a1cbc1, 0xe760d676, 0xea23f0af, 0xeee2ed18,
	0xf0a5bd1d, 0xf464a0aa, 0xf9278673, 0xfde69bc4, 0x89b8fd09,
	0x8d79e0be, 0x803ac667, 0x84fbdbd0, 0x9abc8bd5, 0x9e7d9662,
	0x933eb0bb, 0x97ffad0c, 0xafb010b1, 0xab710d06, 0xa6322bdf,
	0xa2f33668, 0xbcb4666d, 0xb8757bda, 0xb5365d03, 0xb1f740b4
};

/**
 * @brief
 *	-Compute a POSIX 1003.2 checksum.  This routine has been broken out so that
 * 	other programs can use it.  It takes a char pointer and length.
 * 	It ruturns the crc value for the data in buf.
 *
 * @param[in] buf - input data data for which crc computed
 * @param[in] len - length of input data
 *
 * @return	u_long
 * @retval	crc value	success
 * 
 */

u_long
crc(u_char *buf, u_long	clen)
{
	register u_char *p;
	register u_long crc, len;

	#define	COMPUTE(var, ch) (var) = (((var) << 8) ^ \
		crctab[(((var) >> 24) & 0xff) ^ (ch)]) & 0xffffffff

	for (crc=0, len=clen, p = buf; len--; ++p) {
		COMPUTE(crc, *p);
	}

	/* Include the length of the file. */
	for (; clen != 0; clen >>= 8) {
		COMPUTE(crc, clen & 0xff);
	}

	return (~crc & 0xffffffff);
}
/*
 *	END of included source
 */

#ifdef WIN32
/**
 * @brief
 * 	Given some bytes of data in 'buf' of size 'buf_sz', return a
 *	copy of the data but with <carriage return> <linefeed> combination
 *	entries replaced by a single <linefeed>. The returned buffer
 *	size is returned in 'new_buf_sz'.
 *
 * @param[in]		buf	- some bytes of data.
 * @param[in]		buf_sz	- size of 'buf'.
 * @param[in/out]	new_buf_sz - holds the buffer size of the returned buf.
 *
 * @return char *
 * @retval <copy of filtered 'buf'>
 */
static char *
dos2unix(char *buf, unsigned long buf_sz, int *new_buf_sz)
{
	static char *buf2 = NULL;
	static	unsigned long buf2_sz = 0;
	char	*tmp_str = NULL;
	int	i, j;

	if (buf_sz > buf2_sz) {
		tmp_str = realloc(buf2, buf_sz);
		if (tmp_str == NULL) {
			*new_buf_sz = buf_sz;
			return (buf); /* return original */
		}
		buf2 = tmp_str;
		buf2_sz = buf_sz;
	}

	memset(buf2, '\0', buf2_sz);
	j = 0;
	for (i=0; i < buf_sz; i++) {
		if ((i < (buf_sz-1)) && (buf[i] == '\r') &&
			(buf[i+1] == '\n')) {
			buf2[j++] =  '\n';
			i++; /* skip the next linefeed */
		} else {
			buf2[j++] = buf[i];
		}
	}

	*new_buf_sz = j;
	return (buf2);
}
#endif

/**
 * @brief
 * 	Given a file represented by 'filepath', return its crc value. 
 *
 * @param[in]	filepath	- file being crc-ed.
 *
 * @return u_long
 * @retval	> 0	- crc (checksum) value of the file.
 * @retval	0	- if file is non-existent, or file is empty, or if an
 * 			  error encountered while opening or reading the
 * 			  file.
 */
unsigned long
crc_file(char *filepath)
{
	int     fd;
	struct stat sb;
	static u_char *buf = NULL;
	static	int buf_sz = 0;
	u_char	*tmp_str = NULL;
	int	nread = 0;
	int	count;
	u_char	*tmpbuf;	
#ifdef WIN32
	u_char	*tr_buf = NULL;
	int	tr_buf_sz = 0;
#endif


	if (filepath == NULL)
		return (0);
	

	if (stat(filepath, &sb) == -1) {
		return (0);
	}

	if (sb.st_size <= 0) {
		return (0);
	}

	if ((fd = open(filepath, O_RDONLY)) <= 0) {
		return (0);
	}

#ifdef WIN32
	setmode(fd, O_BINARY);
#endif

	if (sb.st_size > buf_sz) {
		tmp_str = realloc(buf, sb.st_size);
		if (tmp_str == NULL) {
			close(fd);
			return (0);
		}
		buf = tmp_str;
		buf[0] = '\0';
		buf_sz = sb.st_size;
	}

	tmpbuf = buf;
	count = sb.st_size;

	while ( ((nread=read(fd, tmpbuf, count)) > 0) &&
					(nread <= sb.st_size) ) {

		count -= nread;
		tmpbuf += nread;

		if (count == 0) {
			break;
		}
	}

	if (nread < 0) {
		close(fd);
		return (0);
	}

	close(fd);
#ifdef WIN32
	tr_buf = dos2unix(buf, sb.st_size, &tr_buf_sz);
	return (crc(tr_buf, tr_buf_sz));
#else
	return (crc(buf, sb.st_size));
#endif
}

/**
 * @brief
 *	-Generate a sequence number for a packet.
 *
 * @param[in] seq - previous sequence number
 * 
 * @return	int
 * @retval	0	success
 * @retval	-1	error
 *
 */
static int
next_seq(int *seq)
{
	(*seq)++;
	if (*seq < 0) {		/* had a rollover */
		errno = EFBIG;
		return -1;
	}

	return 0;
}

/**
 * @brief
 *	Put a human readable representation of a network addres into
 *	a staticly allocated string.
 *
 * @param[in] ap - internet address
 *
 * @return	string
 * @retval	static  string		success
 * @retval	"unknown"		error
 *
 */
char *
netaddr(struct sockaddr_in *ap)
{
	static	char	out[80];
	u_long		ipadd;

	if (ap == NULL)
		return "unknown";

	ipadd = ntohl(ap->sin_addr.s_addr);

	sprintf(out, "%ld.%ld.%ld.%ld:%d",
		(ipadd & 0xff000000) >> 24,
		(ipadd & 0x00ff0000) >> 16,
		(ipadd & 0x0000ff00) >> 8,
		(ipadd & 0x000000ff),
		ntohs(ap->sin_port));
	return out;
}

#ifdef WIN32

#ifndef SIO_UDP_CONNRESET
#define SIO_UDP_CONNRESET _WSAIOW(IOC_VENDOR, 12)
#endif

/**
 * @brief
 * 	-fix_connreset_msbug: this is the Microsoft fix to issue Q263823 where 
 *	if sendto(sd) receives an "ICMP Port unreachable" message,  then 
 *      a corresponding select() and a recvfrom() will return 
 *	"WSAECONNRESET (10054)" instead of blocking or timing out. 
 *      NOTE: This fix requires Windows 2000 Service Pack 2 
 *
 */

static void
fix_connreset_msbug(int sd)
{
	DWORD	dwBytesReturned = 0;
	BOOL	bNewBehavior = FALSE;

	/* disable  new behavior using */
	/* IOCTL: SIO_UDP_CONNRESET */

	WSAIoctl(sd, SIO_UDP_CONNRESET, &bNewBehavior, sizeof(bNewBehavior),
		NULL, 0, &dwBytesReturned, NULL, NULL);
}
#endif	/* WIN32 */

/**
 * @brief
 *	-Create a packet of the given type, fill in the sequence and
 *	index number.  If buf is NULL, malloc an area for just
 *	a header.  If buf is not NULL, it should contain space
 *	for len+RPP_PKT_HEADER bytes.
 *
 * @param[in] index - index number
 * @param[in] type -type of packet
 * @param[in] seq - sequence number
 * @param[in] buf - buffer to hold packet data
 * @param[in] len - length of buf
 *
 * @return	Void
 *
 */
static void
__rpp_form_pkt(int index, int type, int seq, u_char *buf, int len)
{
	struct	send_packet	*pktp;
	struct	stream		*sp;

	DBPRT((DBTO, "%s: index %d type %d seq %d len %d\n",
		__func__, index, type, seq, len))
	sp = &stream_array[index];
	pktp = (struct send_packet *)malloc(sizeof(struct send_packet));
	assert(pktp != NULL);

	pktp->type = type;
	pktp->sequence = seq;
	pktp->time_sent = 0;
	pktp->first_sent = 0;
	pktp->retries = 0;
	pktp->len = len;
	pktp->index = index;

	if (buf)
		pktp->data = (u_char *)realloc(buf, len+RPP_PKT_HEAD);
	else
		pktp->data = (u_char *)malloc(RPP_PKT_HEAD);
	assert(pktp->data != NULL);
	/*
	 ** Put on stream send list
	 */
	if (sp->send_head == NULL)
		sp->send_head = pktp;
	else
		sp->send_tail->next = pktp;
	sp->send_tail = pktp;
	pktp->next = NULL;
	pktp->down = NULL;

	/*
	 ** if open has not completed yet, hold off putting on send queue
	 */
	if (sp->stream_id == -1) {
		pktp->up = NULL;
		return;
	}
	/*
	 ** if the stream is fully open, format and put on the send queue
	 */
	DBPRT((DBTO, "%s: idx %d link %d seq %d len %d to sendq\n",
		__func__, index, type, seq, len))

	I2TOH(type, (char *)&pktp->data[len])
	I8TOH(sp->stream_id, (char *)&pktp->data[len+RPP_HDR_SID])
	I8TOH(seq, (char *)&pktp->data[len+RPP_HDR_SEQ])
	I8TOH(crc(pktp->data, (u_long)(len+RPP_PKT_CRC)),
		(char *)&pktp->data[len+RPP_PKT_CRC])

	if (bottom)
		bottom->down = pktp;
	pktp->up = bottom;
	if (top == NULL)		/* first one */
		top = pktp;
	bottom = pktp;
	return;
}

/**
 * @brief
 *	-Check to make sure an incoming packet goes with one of the
 *	streams we have.
 *
 * @param[in] index - index number
 * @param[in] addrp - internet address
 *
 * @return	structure handle
 * @retval	pointer to stream	success
 * @retval	NULL			error
 *
 */
static
struct	stream	*
__rpp_check_pkt(int index, struct	sockaddr_in *addrp)
{
	struct	stream	*sp;
	struct	in_addr	*addrs;
	int		i;

	if (index < 0 || index >= stream_num) {
		DBPRT((DBTO, "%s: BAD INDEX %d outside limit %d\n",
			__func__, index, stream_num))
		return NULL;
	}

	sp = &stream_array[index];
	if (sp->state <= RPP_FREE) {
		DBPRT((DBTO, "%s: FREE STREAM\n", __func__))
		return NULL;
	}

	if (addrp->sin_port != sp->addr.sin_port)
		goto bad;
	if (addrp->sin_family != sp->addr.sin_family)
		goto bad;
	if (addrp->sin_addr.s_addr == sp->addr.sin_addr.s_addr)
		return sp;

	if ((addrs = sp->addr_array) != NULL) {
		for (i=0; addrs[i].s_addr; i++) {
			if (addrs[i].s_addr == addrp->sin_addr.s_addr)
				return sp;
		}
	}

bad:
	DBPRT((DBTO, "%s: ADDRESS MISMATCH\n", __func__))
	DBPRT((DBTO, "\tstream %d addr %s\n", index, netaddr(&sp->addr)))
	DBPRT((DBTO, "\tpkt addr %s\n", netaddr(addrp)))
	return NULL;
}

/**
 * @brief
 *	-Send outstanding information starting with top and working
 *	down to bottom.  Will not cause state change.
 */
static
void
__rpp_send_out(void)
{
	struct	send_packet	*pp;
	struct	stream		*sp;
	time_t	curr = time(NULL);
	static	time_t	error_time = 0;
	static	time_t	send_hold;
	static	time_t	min_hold = 2;	/* seconds */
	static	time_t	max_hold = 8;	/* seconds */

	/*
	 ** Check if logging is being done and it is time to print
	 ** a message.  Don't print anything if there have been no
	 ** retries or fails in the last reporting period.
	 */
	if (rpp_logfunc != NULL && curr > (last_time + RPP_LOG_PERIOD)) {
		if (last_fails > 0 || last_retries > 0 || last_errors > 0) {
			if (last_fails > 0 || last_retries > 0) {
				sprintf(rpp_logbuf,
					"total (pkts=%lu,retries=%lu,fails=%lu) "
					"last %d secs (pkts=%lu,retries=%lu,fails=%lu)",
					total_packets, total_retries, total_fails,
					(int)(curr - last_time),
					last_packets, last_retries, last_fails);
				rpp_logfunc(rpp_logbuf);
				last_packets = last_retries = last_fails = 0;
			}
			if (last_errors > 0) {
				sprintf(rpp_logbuf,
					"errors total %lu last %d secs %lu",
					total_errors, (int)(curr - last_time), last_errors);
				rpp_logfunc(rpp_logbuf);
				last_errors = 0;
			}
			last_time = curr;
		}
	}

	/* check if we are in a timeout period for sending packets */
	if ((error_time != 0) && (error_time + send_hold > curr))
		return;

	for (pp = top; pp; pp = pp->down) {
		/*
		 ** Check if this packet has never been sent before.
		 */
		if (pp->time_sent == 0) {
			/*
			 ** Don't exceed rpp_highwater total packets
			 ** "on the wire".
			 */
			if (pkts_sent >= rpp_highwater)
				continue;
		}
		else {
			/*
			 ** Don't resend a packet more than rpp_retry times.
			 */
			if (pp->retries >= rpp_retry)
				continue;
			/*
			 ** Don't send the same packet too often.
			 */
			if ((curr - pp->time_sent) < RPP_TIMEOUT)
				continue;
		}

		sp = &stream_array[pp->index];
		DBPRT((DBTO,
			"%s index %d type %d retry %d seq %d to %s crc %8.8s\n",
			__func__, pp->index, pp->type,
			pp->retries, pp->sequence, netaddr(&sp->addr),
			(char *)&pp->data[pp->len+RPP_PKT_CRC]))
		if (sendto(sp->fd, (char *)pp->data, RPP_PKT_HEAD+pp->len,
				0, (struct sockaddr *)&sp->addr,
				sizeof(struct sockaddr_in)) == -1) {
			total_errors++;
			last_errors++;
			if (rpp_logfunc != NULL) {
				snprintf(rpp_logbuf, sizeof(rpp_logbuf),
					"RPP sendto error %d, %s", errno, strerror(errno));
				rpp_logfunc(rpp_logbuf);
			}
			/* Check if the error is one that should result in not
			 * sending packets for awhile.  If so, calculate how long
			 * the time should be.  If this is the first send try after
			 * a previous error, increase the wait time up to max_hold.
			 * If it is a different error, treat it as if it worked to
			 * allow the highwater/retry counts to work.
			 */
			if (errno == ENOBUFS) {
				error_time = curr;
				if (send_hold == 0)
					send_hold = min_hold;
				else {
					send_hold *= 2;
					if (send_hold > max_hold)
						send_hold = max_hold;
				}
				return;
			}
		}

		if (pp->time_sent == 0) {		/* new one */
			pp->first_sent = curr;
			pkts_sent++;
		}
		else {
			total_retries++;
			last_retries++;
			pp->retries++;
		}

		total_packets++;
		last_packets++;
		error_time = 0;
		send_hold = 0;

		pp->time_sent = curr;
	}

	return;
}

/**
 * @brief
 *	Create or reuse a position in stream_array.
 *
 * @return 	int
 * @retval 	>= 0	stream number created/reused
 * @retval 	-1	error if stream creation/reuse fails.
 */
static
int
__rpp_create_sp(void)
{
	int	i;
	struct	stream	*sp = NULL;

	if (stream_array == NULL) {
		stream_array = (struct stream *)malloc(sizeof(struct stream));
		if (stream_array == NULL)
			return -1;
		memset(stream_array, '\0', sizeof(struct stream));
		stream_num = 1;
	}
	for (i=0; i<stream_num; i++) {
		sp = &stream_array[i];

		if (sp->state == RPP_FREE)
			break;
	}
	if (i == stream_num) {
		for (i=0; i<stream_num; i++) {
			sp = &stream_array[i];

			if (sp->state == RPP_DEAD)
				break;
		}
	}
	if (i == stream_num) {		/* none available */
		sp = (struct stream *)realloc((void *)stream_array,
			(stream_num*2)*sizeof(struct stream));
		if (sp == NULL) {
			sp = (struct stream *)realloc((void *)stream_array,
				(stream_num+1)*sizeof(struct stream));
			if (sp == NULL)
				return -1;
			stream_num++;
		}
		else
			stream_num *= 2;
		stream_array = sp;
		sp = &stream_array[i];
		memset((void *)sp, '\0', (stream_num-i)*sizeof(struct stream));
	}
	else
		memset((void *)sp, '\0', sizeof(struct stream));
	DBPRT((DBTO, "__rpp_create_sp: new index %d\n", i))
	return i;
}

/**
 * @brief
 *	-returns error number specific to host.
 *
 * @param[in] then - time which is used to retry
 *
 * @return	int
 * @retval	error number	success
 *
 */
static
int
host_err(time_t then)
{
	int	ret = 0;

	switch (h_errno) {

		case HOST_NOT_FOUND:
			ret = errno = ENOENT;
			break;

		case TRY_AGAIN:		/* temp error */
			if ((time(NULL) - then) >= 2)	/* time to quit */
				ret = errno = EAGAIN;
			break;

		case NO_RECOVERY:
			ret = errno = ENOEXEC;
			break;

		case NO_DATA:
			ret = errno = EINVAL;
			break;

		default:
			if (errno == 0)
				errno = ENXIO;
			ret = errno;
			break;
	}
	return ret;
}

/**
 * @brief
 *	-returns the host detail of host with name "name".
 *
 * @param[in] name - host name
 *
 * @return	structure handle
 * @retval	pinter to hostentry	success
 * @retval	NULL			error
 *
 */
static
struct  hostent *
hostbyname(char *name)
{
	struct	hostent		*hp;
	time_t	now = time(NULL);

	errno = 0;
	for (;;) {
		hp = gethostbyname(name);
		if (hp != NULL)
			break;
		if (host_err(now))
			break;
	}
	return hp;
}

/**
 * @brief
 *	-returns the host detail of host with address "addr".
 *
 * @param[in] addr - ip address
 * @param[in] len - length of address
 * @param[in] type - type of address
 *
 * @return      structure handle
 * @retval      pinter to hostentry     success
 * @retval	NULL			error
 *
 */
static
struct  hostent *
hostbyaddr(const char *addr, int len, int type)
{
	struct	hostent		*hp;
	time_t	now = time(NULL);

	errno = 0;
	for (;;) {
		hp = gethostbyaddr(addr, len, type);
		if (hp != NULL)
			break;
		if (host_err(now))
			break;
	}
	return hp;
}

/**
 * @brief
 *	-Look up the "canonical" name for the host by
 *	calling gethostbyaddr with an IP address.
 *
 * @param[in] addr - pointer to socket fields
 *
 * @return	 structure handle
 * @retval      pinter to hostentry     success
 * @retval      NULL                    error
 *
 */
static
struct	hostent		*
__rpp_get_cname(struct sockaddr_in *addr)
{
	struct	hostent		*hp;
	char			*hname;

	if ((hp = hostbyaddr((void *)&addr->sin_addr,
		sizeof(struct in_addr),
		addr->sin_family)) == NULL) {
		DBPRT((DBTO, "%s: addr not found, h_errno=%d errno=%d\n",
			__func__, h_errno, errno))
		return NULL;
	}
	if ((hname = (char *)strdup(hp->h_name)) == NULL)
		return NULL;

	if ((hp = hostbyname(hname)) == NULL) {
		DBPRT((DBTO,
			"%s: canonical name %s not found, h_errno=%d errno=%d\n",
			__func__, hname, h_errno, errno))
	}
	free(hname);
	return hp;
}

/**
 * @brief
 *	-Allocate a list of alternate address for a host and save
 *	them in the stream structure.
 *
 * @param[in] hp - pointer to host list
 * @param[in] sp - pointer to stream
 *
 * @return Void
 *
 */
static
void
__rpp_alist(struct hostent *hp, struct stream *sp)
{
	int	i, j;

	if (!hp || !sp)
		return;

	for (i=1; hp->h_addr_list[i]; i++);
	if (i == 1)
		return;

	sp->addr_array = (struct in_addr *)calloc(i, sizeof(struct in_addr));
	if (sp->addr_array) {
		for (j=i=0; hp->h_addr_list[i]; i++) {
			if (memcmp(&sp->addr.sin_addr,
				hp->h_addr_list[i], hp->h_length) == 0)
				continue;
			memcpy(&sp->addr_array[j++], hp->h_addr_list[i], hp->h_length);
		}
		sp->addr_array[j].s_addr = 0;
	} else
		pbs_errno = PBSE_SYSTEM;

	return;
}

/**
 * @brief
 *	-send acknowledgement for rpp .
 *
 * @param[in] sp - pointer to stream
 * @param[in] seq - sequence number for ackn
 *
 * @return	int
 * @retval	0	success
 * @retval	-1	error
 *
 */
static
int
__rpp_send_ack(struct stream *sp, int seq)
{
	char	buf[RPP_PKT_HEAD];
	u_long	xcrc;

	if (sp->stream_id < 0) {		/* can't send yet */
		DBPRT((DBTO, "%s: STREAM NOT OPEN seq %d\n", __func__, seq))
		return 0;
	}

	I2TOH(RPP_ACK, buf)
	I8TOH(sp->stream_id, &buf[2])
	I8TOH(seq, &buf[10])
	xcrc = crc((u_char *)buf, (u_long)RPP_PKT_CRC);
	I8TOH(xcrc, &buf[RPP_PKT_CRC])

	DBPRT((DBTO, "%s: seq %d to %s crc %lX\n",
		__func__, seq, netaddr(&sp->addr), xcrc))
	if (sendto(sp->fd, buf, RPP_PKT_HEAD, 0, (struct sockaddr *)&sp->addr,
		sizeof(struct sockaddr_in)) == -1) {
		DBPRT((DBTO, "%s: ACK error %d\n", __func__, errno))
		if (errno != EWOULDBLOCK && errno != ENOBUFS)
			return -1;
	}
	return 0;
}

/**
 * @brief
 *	-Take a packet off the send queue and free it.
 *
 * @param[in] pp - pointer to packet
 *
 * @return	Void
 *
 */
static
void
dqueue(struct send_packet *pp)
{
	if (pp->down == NULL)
		bottom = pp->up;
	else
		pp->down->up = pp->up;
	if (pp->up == NULL)
		top = pp->down;
	else
		pp->up->down = pp->down;

	if (--pkts_sent < 0)
		pkts_sent = 0;
	if (pp->data)
		free(pp->data);
	free(pp);
	return;
}

/**
 * @brief
 *	-Get rid of anything on the pend and send queue for a stream.
 *
 * @param[in] sp - pointer to stream
 *
 * @return Void
 *
 */
static
void
clear_send(struct stream *sp)
{
	struct	pending		*ppp, *pprev;
	struct	send_packet	*spp, *sprev;

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

	for (spp=sp->send_head; spp; spp=sprev) {
		sprev=spp->next;

		if (sp->stream_id == -1) { 	    	/* not open yet */
			struct	send_packet	*look;	/* might not be */
			/* on send queue */
			for (look = top; look; look = look->down) {
				if (look == spp)
					break;
			}
			if (look == NULL) {
				if (spp->data)
					free(spp->data);
				free(spp);
				continue;
			}
		}
		dqueue(spp);
	}
	sp->send_head = NULL;
	sp->send_tail = NULL;
}

/**
 * @brief
 *	-Remove packets from receive, pending and send queues for
 *	a stream, free all the memory and zero the stream_array
 *	entry.
 *
 * @param[in] sp - pointer to stream
 *
 * @return Void
 *
 */

static
void
clear_stream(struct stream *sp)
{
	struct	recv_packet	*rpp, *rprev;

	DBPRT((DBTO, "CLEAR stream %ld\n",
		((long)sp - (long)stream_array)/sizeof(struct stream)))
	for (rpp=sp->recv_head; rpp; rpp=rprev) {
		rprev=rpp->next;
		if (rpp->data)
			free(rpp->data);
		free(rpp);
	}
	sp->recv_head = NULL;
	sp->recv_tail = NULL;

	clear_send(sp);
	if (sp->addr_array) {
		free(sp->addr_array);
		sp->addr_array = NULL;
	}
	sp->state = RPP_DEAD;
}

/**
 * @brief
 *	-Do a recvfrom call to get a packet off of all file descriptors.
 *
 * @param[in] fd - file descriptor
 *
 * @return	int
 * @retval	index of the stream the packet belonged to	success
 * @retval	-1						error
 * @retval	-2						not data
 * @retval	-3						no data to read.
 *
 * @par	Side Effects:
 *	MAY CAUSE STATE CHANGE!
 */
static
int
__rpp_recv_pkt(int fd)
{
	ssize_t			len;
	pbs_socklen_t		flen;
	struct	sockaddr_in	addr;
	struct	hostent		*hp;
	int			i, streamid;
	struct	send_packet	*spp, *sprev;
	struct	recv_packet	*rpp, *rprev;
	struct	recv_packet	*pkt;
	struct	stream		*sp;
	char			*data;
	int		type;
	int		sequence;
	u_long		pktcrc;

	data = malloc(RPP_PKT_SIZE);
	assert(data != NULL);
	flen = sizeof(struct sockaddr_in);
	/*
	 **	Loop so we can avoid failing on EINTR.  Thanks to
	 **	Pete Wyckoff for finding this.
	 */
	for (;;) {
#ifdef	_SX
		fd_set			fdset;
		struct	timeval		tv;

		FD_ZERO(&fdset);
		tv.tv_sec = 0;
		tv.tv_usec = 0;

		FD_SET(fd, &fdset);
		i = select(FD_SETSIZE, &fdset, NULL, NULL, &tv);
		if (i == 0) {
			free(data);
			return -3;
		} else if (i == -1) {
			if (errno == EINTR)
				continue;
			free(data);
			return -1;
		}
#endif

		len = recvfrom(fd, data, RPP_PKT_SIZE, 0,
			(struct sockaddr *)&addr, &flen);

		if (len != -1)
			break;
#ifdef WIN32
		errno = WSAGetLastError();
		if (errno == WSAEINTR)
			continue;
#else
		if (errno == EINTR)
			continue;
#endif

		free(data);
		if (errno == EWOULDBLOCK ||
			errno == EAGAIN      ||
			errno == ECONNREFUSED) {
			errno = 0;
			return -3;
		}
		return -1;
	}

	DBPRT((DBTO, "%s: addr %s len %ld\n", __func__, netaddr(&addr), (long)len))

	if (len < RPP_PKT_HEAD)		/* less than minimum size */
		goto err_out;

	HTOI8(&data[len-RPP_CRC_LEN], pktcrc)
	if (pktcrc != crc((u_char *)data, (u_long)(len-RPP_CRC_LEN))) {
		DBPRT((DBTO, "%s: packet crc %08lX failed\n", __func__, pktcrc))
		goto err_out;
	}
	HTOI2(&data[len-RPP_PKT_HEAD], type)
	HTOI8(&data[len-RPP_PKT_HEAD+RPP_HDR_SID], streamid)
	HTOI8(&data[len-RPP_PKT_HEAD+RPP_HDR_SEQ], sequence)

	switch (type) {

		case RPP_ACK:
			DBPRT((DBTO, "%s: ACK stream %d sequence %d crc %08lX\n",
				__func__, streamid, sequence, pktcrc))
			free(data);
			if ((sp = __rpp_check_pkt(streamid, &addr)) == NULL)
				return -2;

			if (sp->state == RPP_OPEN_PEND) {
				if (sequence != sp->open_key) {
					DBPRT((DBTO,
						"%s: WILD ACK in RPP_OPEN_PEND %d\n",
						__func__, streamid))
					return -2;
				}
				spp = sp->send_head;
				assert(spp->type == RPP_HELLO2);
				assert(spp->next == NULL);

				sp->state = RPP_CONNECT;
				sp->send_head = NULL;
				sp->send_tail = NULL;
				dqueue(spp);
				return streamid;
			}
			else if (sp->stream_id == -1) {
				DBPRT((DBTO, "%s: ACK for closed stream %d\n",
					__func__, streamid))
				return -2;
			}

			for (spp=sp->send_head, sprev=NULL; spp;
				sprev=spp, spp=spp->next) {
				if (spp->sequence == sequence)
					break;
			}
			if (spp) {
				DBPRT((DBTO, "%s: stream %d seq %d retry %d took %ld\n",
					__func__, streamid, sequence, (int)spp->retries,
					(long)(time(NULL) - spp->first_sent)))

				if (sp->state == RPP_CLOSE_WAIT1 &&
					spp->type == RPP_GOODBYE)
					sp->state = RPP_CLOSE_WAIT2;

				if (sprev == NULL)
					sp->send_head = spp->next;
				else
					sprev->next = spp->next;
				if (sp->send_tail == spp)
					sp->send_tail = sprev;
				dqueue(spp);

				if (sp->state == RPP_LAST_ACK &&
					sp->send_head == NULL) {
					clear_stream(sp);
					return -2;
				}
			}
			return streamid;

		case RPP_GOODBYE:
			DBPRT((DBTO, "%s: GOODBYE stream %d sequence %d crc %08lX\n",
				__func__, streamid, sequence, pktcrc))
			free(data);
			if ((sp = __rpp_check_pkt(streamid, &addr)) == NULL)
				return -2;
			if (__rpp_send_ack(sp, sequence) == -1)
				return -1;

			switch (sp->state) {

				case RPP_OPEN_PEND:
				case RPP_OPEN_WAIT:
				case RPP_CLOSE_PEND:
				case RPP_LAST_ACK:
					return -2;

				case RPP_CLOSE_WAIT1:
					sp->state = RPP_LAST_ACK;
					return -2;

				case RPP_CLOSE_WAIT2:
					clear_stream(sp);
					return -2;

				default:
					break;
			}

			sp->state = RPP_CLOSE_PEND;
			clear_send(sp);		/* other side not reading now */

			for (rpp=sp->recv_head, rprev=NULL; rpp;
				rprev=rpp, rpp=rpp->next) {
				if (rpp->sequence >= sequence)
					break;
			}
			if (rpp == NULL || rpp->sequence > sequence) {
				DBPRT((DBTO, "%s: GOOD seq %d\n", __func__, sequence))
				pkt = (struct recv_packet *)
					malloc(sizeof(struct recv_packet));
				assert(pkt != NULL);
				pkt->type = type;
				pkt->sequence = sequence;
				pkt->len = 0;
				pkt->data = NULL;
				if (rprev == NULL) {
					pkt->next = sp->recv_head;
					sp->recv_head = pkt;
				}
				else {
					pkt->next = rprev->next;
					rprev->next = pkt;
				}
				if (sp->recv_tail == rprev)
					sp->recv_tail = pkt;
			}
			else {
				DBPRT((DBTO, "%s: DUPLICATE seq %d MAX seen %d\n",
					__func__, sequence, rpp->sequence))
			}
			return -2;

		case RPP_DATA:
		case RPP_EOD:
			DBPRT((DBTO,
				"%s: DATA stream %d sequence %d crc %08lX len %ld\n",
				__func__, streamid, sequence, pktcrc, (long)len))
			if ((sp = __rpp_check_pkt(streamid, &addr)) == NULL)
				goto err_out;
			if (__rpp_send_ack(sp, sequence) == -1) {
				free(data);
				return -1;
			}

			switch (sp->state) {
				case RPP_OPEN_WAIT:
					DBPRT((DBTO,
						"INPUT on unconnected stream %d\n", streamid))
					free(data);
					return -2;
				case RPP_CLOSE_WAIT1:
				case RPP_CLOSE_WAIT2:
				case RPP_LAST_ACK:
					DBPRT((DBTO, "INPUT on closed stream %d\n", streamid))
					free(data);
					return -2;
				default:
					break;
			}

			if (sequence < sp->recv_sequence) {
				DBPRT((DBTO, "%s: OLD seq %d\n", __func__, sequence))
				free(data);
				return -2;
			}

			for (rpp=sp->recv_head, rprev=NULL; rpp;
				rprev=rpp, rpp=rpp->next) {
				if (rpp->sequence >= sequence)
					break;
			}
			if (rpp == NULL || rpp->sequence > sequence) {
				DBPRT((DBTO, "%s: GOOD seq %d\n", __func__, sequence))
				data = realloc(data, len);
				assert(data != NULL);
				pkt = (struct recv_packet *)
					malloc(sizeof(struct recv_packet));
				assert(pkt != NULL);
				pkt->type = type;
				pkt->sequence = sequence;
				pkt->len = len-RPP_PKT_HEAD;
				pkt->data = (u_char *)data;
				if (rprev == NULL) {
					pkt->next = sp->recv_head;
					sp->recv_head = pkt;
				}
				else {
					pkt->next = rprev->next;
					rprev->next = pkt;
				}
				if (sp->recv_tail == rprev)
					sp->recv_tail = pkt;

				if (sp->state == RPP_OPEN_PEND)
					return -2;
				else
					return streamid;
			}
			else {
				DBPRT((DBTO, "%s: DUPLICATE seq %d MAX seen %d\n",
					__func__, sequence, rpp->sequence))
				free(data);
			}
			break;

		case RPP_HELLO1:
			/*
			 ** HELLO1 packets have the remote side's stream index
			 ** in the "streamid" field and open key in the sequence.
			 */
			DBPRT((DBTO, "%s: HELLO1 stream %d sequence %d\n",
				__func__, streamid, sequence))
			free(data);
			for (i=0; i<stream_num; i++) {
				sp = &stream_array[i];

				if (sp->state <= RPP_FREE)
					continue;
				if (memcmp(&sp->addr, &addr, sizeof(addr)))
					continue;
				if (sp->open_key == sequence) {
					__rpp_send_out();
					return -2;
				}
				DBPRT((DBTO, "OLD STREAM state %d reopened %d %d\n",
					sp->state, sp->open_key, sequence))
				clear_stream(sp);	/* old stream */
			}

			i = __rpp_create_sp();
			if (i == -1)
				return -1;
			sp = &stream_array[i];
			sp->state = RPP_OPEN_PEND;
			sp->fd = fd;
			memcpy(&sp->addr, &addr, sizeof(addr));
			/*
			 * Need not check canonical name for local host
			 * alternate address list will always include loopback address
			 * as the alternatye address.
			 */

			if ((hp = __rpp_get_cname(&addr)) != NULL)
				__rpp_alist(hp, sp);
			sp->stream_id = streamid;
			sp->open_key = sequence;
			open_key = MAX(open_key, sequence);
			__rpp_form_pkt(i, RPP_HELLO2, i, NULL, 0);
			__rpp_send_out();
			break;

		case RPP_HELLO2:
			/*
			 ** HELLO2 packet has this side's stream index in
			 ** "streamid" as usual and the remote side's
			 ** stream index overloaded in the "sequence" field.
			 */
			DBPRT((DBTO, "%s: HELLO2 stream %d sequence %d\n",
				__func__, streamid, sequence))
			free(data);
			if ((sp = __rpp_check_pkt(streamid, &addr)) == NULL)
				return -2;

			switch (sp->state) {

				case RPP_OPEN_WAIT:
					sp->state = RPP_CONNECT;
					break;

				case RPP_CLOSE_WAIT1:	/* called close before open done */
				case RPP_LAST_ACK:
					break;

				default:
					if (sp->stream_id == sequence) {
						DBPRT((DBTO,
							"%s: stream %d got DUP HELLO2 %d\n",
							__func__, streamid, sp->state))
						if (__rpp_send_ack(sp, sp->open_key) == -1)
							return -1;
					}
					else {
						DBPRT((DBTO, "%s: NON-DUP HELLO2\n", __func__))
					}
					return -2;
			}

			sp->stream_id = sequence;
			if (__rpp_send_ack(sp, sp->open_key) == -1)
				return -1;

			if ((spp = sp->send_head) == NULL) {
				DBPRT((DBTO,
					"%s: stream %d got HELLO2 but sendq NULL\n",
					__func__, streamid))
				return -2;
			}
			if (spp->type != RPP_HELLO1) {
				DBPRT((DBTO,
					"%s: stream %d sendq %d rather than HELLO1\n",
					__func__, streamid, spp->type))
				return -2;
			}
			sp->send_head = spp->next;	/* remove HELLO1 pkt */
			if (sp->send_tail == spp)
				sp->send_tail = NULL;
			dqueue(spp);

			/*
			 ** Put any waitting packets onto the send queue
			 */
			for (spp=sp->send_head; spp; spp=spp->next) {
				int	len = spp->len;

				DBPRT((DBTO,
					"%s: idx %d link %d seq %d len %d to sendq\n",
					__func__, streamid, spp->type, spp->sequence, len))
				I2TOH(spp->type, (char *)&spp->data[len])
				I8TOH(sp->stream_id,
					(char *)&spp->data[len+RPP_HDR_SID])
				I8TOH(spp->sequence,
					(char *)&spp->data[len+RPP_HDR_SEQ])
				I8TOH(crc(spp->data, (u_long)(len+RPP_PKT_CRC)),
					(char *)&spp->data[len+RPP_PKT_CRC])

				if (bottom)
					bottom->down = spp;
				spp->up = bottom;
				spp->down = NULL;
				if (top == NULL)		/* first one */
					top = spp;
				bottom = spp;
			}

			break;

		default:
			DBPRT((DBTO,
				"%s: UNKNOWN packet type %d stream %d sequence %d\n",
				__func__, type, streamid, sequence))
			free(data);
			break;
	}
	return -2;

err_out:
	free(data);
	return -2;
}

/**
 * @brief
 *	-Do recv calls until there is one that shows data.
 *
 * @return      int
 * @retval      index of the stream the packet belonged to      success
 * @retval      -1                                              error
 * @retval      -2                                              not data
 * @retval      -3                                              no data to read.
 * 
 */
static
int
__rpp_recv_all(void)
{
	int	i, ret;
	int	rc = -3;

	for (i=0; i<rpp_fd_num; i++) {
		ret = __rpp_recv_pkt(rpp_fd_array[i]);
		rc = MAX(ret, rc);
		if (ret == -1)
			break;
	}
	return rc;
}

/**
 * @brief
 *	Check to see if any packet being sent out on a stream has
 *	been sent more than a reasonable number of times.
 *
 * @param[in] sp - pointer to stream
 *
 */
static
void
__rpp_stale(struct stream *sp)
{
	struct	send_packet	*pp;
	time_t	curr;

	if (sp->state <= RPP_FREE || sp->state == RPP_STALE)
		return;

	curr = time(NULL);
	for (pp = sp->send_head; pp; pp = pp->next) {
		if (pp->time_sent != 0 && pp->retries >= rpp_retry &&
			(curr - pp->time_sent) > RPP_TIMEOUT)
			break;
	}
	if (pp) {
		total_fails++;
		last_fails++;

		DBPRT((DBTO, "STALE PACKET seq %d retry %d of %d\n",
			pp->sequence, pp->retries, rpp_retry))
		switch (sp->state) {
			case RPP_OPEN_PEND:
			case RPP_CLOSE_WAIT1:
			case RPP_CLOSE_WAIT2:
			case RPP_LAST_ACK:
				clear_stream(sp);
				break;
			default:
				sp->state = RPP_STALE;
				break;
		}
	}
}

/**
 * @brief
 *	-Form data packets for any pending data.  If flag is true,
 *	create an EOD packet too.
 *
 * @param[in] index - index of data packet
 * @param[in] flag - indication to create EOD packet or not
 *
 * @return	int
 * @retval	0	success
 * @retval	-1	error
 *
 */
static
int
__rpp_dopending(int index, int flag)
{
	struct	stream		*sp;
	struct	pending		*pp;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, index))
	sp = &stream_array[index];

	for (pp=sp->pend_head; pp != sp->pend_tail; pp=sp->pend_head) {
		__rpp_form_pkt(index, RPP_DATA, sp->send_sequence,
			pp->data, RPP_PKT_DATA);
		sp->pend_head = pp->next;
		free(pp);
		sp->pend_attempt -= RPP_PKT_DATA;
		if (next_seq(&sp->send_sequence) == -1)
			return -1;
	}
	if (flag) {
		__rpp_form_pkt(index, RPP_EOD, sp->send_sequence,
			pp ? pp->data : NULL, sp->pend_attempt);
		if (pp) {
			free(pp);
			sp->pend_head = NULL;
			sp->pend_tail = NULL;
		}
		sp->pend_attempt = 0;
		if (next_seq(&sp->send_sequence) == -1)
			return -1;
	}
	sp->pend_commit = sp->pend_attempt;
	return 0;
}

/**
 * @brief
 *	-Flush all data out of a stream -- do an end of message.
 *
 * @param[in] index - index of data
 *
 * @return	int
 * @retval	0	if it all went well
 * @retval	-1	error
 *
 */
int
__rpp_flush(int index)
{
	struct	stream	*sp;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, 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_DEAD:
		case RPP_FREE:
		case RPP_OPEN_PEND:
		case RPP_CLOSE_WAIT1:
		case RPP_CLOSE_WAIT2:
		case RPP_LAST_ACK:
			errno = ENOTCONN;
			return -1;

		default:
			break;
	}

	/**
	 *	if something is pending or we need to return a zero len EOM,
	 *	call __rpp_dopending().
	 */
	errno = 0;
	if (sp->pend_head != NULL || sp->send_head == NULL) {
		if (__rpp_dopending(index, TRUE))
			return -1;
	}

	if (__rpp_recv_all() == -1)
		return -1;
	__rpp_send_out();
	return 0;
}

/**
 * @brief
 *	-Create a new socket if needed and bind a local port.
 *	If port is 0, pick a free port number.
 *
 * @param[in] port - local port number
 *
 * @return	int
 * @retval	socket fd	success
 * @retval	-1		error
 *
 */
int
__rpp_bind(uint port)
{
	struct	sockaddr_in	from;
#ifdef WIN32
	unsigned long 		num_bytes = 1;
#else
	int			flags;
	socklen_t		size;
#endif

	if (last_time == 0)
		last_time = time(NULL);

	if (rpp_fd == -1) {
		if ((rpp_fd = socket(PF_INET, SOCK_DGRAM, 0)) == -1)
			return -1;

#ifdef WIN32
		fix_connreset_msbug(rpp_fd);

		/* set no delay - windows 2000 port version */
		if ((ioctlsocket(rpp_fd, FIONBIO, &num_bytes)) == SOCKET_ERROR) {
			closesocket(rpp_fd);
			rpp_fd = -1;
			return -1;
		}

#else	/* UNIX code here */
		/* set close on exec */
		if ((flags = fcntl(rpp_fd, F_GETFD)) == -1) {
			close(rpp_fd);
			rpp_fd = -1;
			return -1;
		}
		flags |= FD_CLOEXEC;
		if (fcntl(rpp_fd, F_SETFD, flags) == -1) {
			close(rpp_fd);
			rpp_fd = -1;
			return -1;
		}

		/* set no delay */
		if ((flags = fcntl(rpp_fd, F_GETFL)) == -1) {
			close(rpp_fd);
			rpp_fd = -1;
			return -1;
		}
#if defined(FNDELAY)
		flags |= FNDELAY;
#else
		flags |= O_NONBLOCK;
#endif
		if (fcntl(rpp_fd, F_SETFL, flags) == -1) {
			close(rpp_fd);
			rpp_fd = -1;
			return -1;
		}
		/*
		 * 1. If rpp_cur_rcvbuf_size is initialized, then use it to
		 *	set SO_RCVBUF value. Otherwise,
		 * 2. Try to set SO_RCVBUF in a loop, starting with
		 *  rpp_max_rcvbuf_size.
		 *   2.a. Half the rpp_cur_rcvbuf_size each time
		 *		 setsockopt fails.
		 *   2.b. Exit from the loop when rpp_cur_rcvbuf_size value goes
		 *	 below rpp_min_rcvbuf_size value or 'setsockopt' works.
		 * 3. return -1, if unable to set SO_RCVBUF value.
		 */
		size = sizeof(rpp_cur_rcvbuf_size);
		if (rpp_cur_rcvbuf_size != -1) {
			if (setsockopt(rpp_fd, SOL_SOCKET, SO_RCVBUF,
				&rpp_cur_rcvbuf_size, size)) {
				close(rpp_fd);
				rpp_cur_rcvbuf_size = -1;
				rpp_fd = -1;
				return -1;
			}
		} else {
			for (rpp_cur_rcvbuf_size = rpp_max_rcvbuf_size;
				rpp_cur_rcvbuf_size >= rpp_min_rcvbuf_size;
				rpp_cur_rcvbuf_size /= 2) {
				if (setsockopt(rpp_fd, SOL_SOCKET, SO_RCVBUF,
					&rpp_cur_rcvbuf_size, size) == 0)
					break;
			}
			if (rpp_cur_rcvbuf_size < rpp_min_rcvbuf_size) {
				/* Try to set to rpp_min_rcvbuf_size */
				rpp_cur_rcvbuf_size = rpp_min_rcvbuf_size;
				if (setsockopt(rpp_fd, SOL_SOCKET, SO_RCVBUF,
					&rpp_cur_rcvbuf_size, size)) {
					close(rpp_fd);
					rpp_cur_rcvbuf_size = -1;
					rpp_fd = -1;
					return -1;
				}
			}
		}

#endif /* end of ifdef win32 */

		/* save who started it all for __rpp_shutdown */
		bind_pid = (pid_t)getpid();
	}

	if (rpp_fd_array != NULL) {
		int	i;

		for (i=0; i<rpp_fd_num; i++) {
			if (rpp_fd_array[i] == rpp_fd)
				return rpp_fd;
		}
	}
	memset(&from, '\0', sizeof(from));
	from.sin_family = AF_INET;

	from.sin_addr.s_addr = htonl(INADDR_ANY);
	from.sin_port = htons((u_short)port);

	if (bind(rpp_fd, (struct sockaddr *)&from, sizeof(from)) == -1) {
#ifdef WIN32
		errno = WSAGetLastError();
#endif
		return -1;
	}

	DBPRT((DBTO, "bind to port %d\n", ntohs(from.sin_port)))
	errno = 0;
	if (rpp_fd_array == NULL) {
		rpp_fd_array = (int *)malloc(sizeof(int));
		rpp_fd_num = 1;
		(void)atexit(__rpp_shutdown);
	}
	else {
		rpp_fd_num++;
		rpp_fd_array = (int *)realloc(rpp_fd_array,
			sizeof(int)*rpp_fd_num);
	}
	assert(rpp_fd_array);
	rpp_fd_array[rpp_fd_num-1] = rpp_fd;
	return rpp_fd;
}

/**
 * @brief
 *	Allocate a communication stream.
 *
 * @param[in]   name - the name of the host(node).
 * @param[in]   port - port number on which to connect
 *
 * @return int
 * @retval >= 0	the rpps stream number
 * @retval -1	error opening rpp stream
 */

int
__rpp_open(char *name, uint port)
{
	int			i, stream;
	struct	hostent		*hp;
	struct	stream		*sp;
#ifdef WIN32
	struct sockaddr_in	*host_addr;
#endif

	DBPRT((DBTO, "%s: entered %s:%d\n", __func__, name, port))
	errno = 0;

	if (__rpp_bind(0) == -1)	/* bind if we need to */
		return -1;

	/*
	 ** First, we look up the IP address for this name.
	 */
	if ((hp = hostbyname(name)) == NULL) {
		DBPRT((DBTO, "%s: host %s not found\n", __func__, name))
			errno = ENOENT;
		return -1;
	}

	/*
	 **	Look for previously existant stream to the given
	 **	host.  If one is found in an open state, just
	 **	return it.
	 */
	for (i=0; i<stream_num; i++) {
		sp = &stream_array[i];
		if (sp->state <= RPP_FREE)
			continue;
		if (memcmp(&sp->addr.sin_addr, hp->h_addr, hp->h_length))
			continue;
		if (sp->addr.sin_port != htons((unsigned short)port))
			continue;
		if (sp->addr.sin_family != hp->h_addrtype)
			continue;
		if (sp->state > RPP_CLOSE_PEND) {
			DBPRT((DBTO, "%s: OLD STREAM state %d reopened %d\n",
				__func__, sp->state, sp->open_key))

			clear_stream(sp);	/* old stream */
		}
		else {
			DBPRT((DBTO,
				"%s: reopen of %s, rpp_retry %d\n",
				__func__, netaddr(&sp->addr), rpp_retry))
			return i;
		}
	}

	stream = __rpp_create_sp();
	if (stream == -1)
		return -1;
	sp = &stream_array[stream];

	if (open_key == 0)
		open_key = (int)time(0) & 0x0fff;
	/*
	 ** We save the address returned for the name given so we
	 ** can send out on the perfered interface.
	 */
	memcpy(&sp->addr.sin_addr, hp->h_addr, hp->h_length);
	sp->addr.sin_port = htons((unsigned short)port);
	sp->addr.sin_family = hp->h_addrtype;
	sp->fd = rpp_fd;
	if (hp->h_addr_list[1] == NULL) {
		if ((hp = __rpp_get_cname(&sp->addr)) == NULL) {
			errno = ENOENT;
			return -1;
		}
	}
	__rpp_alist(hp, sp);
	sp->stream_id = stream;	/* use my streamid for HELLO1 */
	sp->state = RPP_OPEN_WAIT;
	sp->open_key = open_key++;
	__rpp_form_pkt(stream, RPP_HELLO1, sp->open_key, NULL, 0);
	sp->stream_id = -1;	/* don't know his stream id yet */

	if (__rpp_recv_all() == -1)
		return -1;
	__rpp_send_out();
	return stream;
}

/**
 * @brief
 *	-return the network address for a stream.
 *
 * @param[in] index - index of data
 *
 * @return	structure handle
 * @retval	network address		success
 * @retval	NULL			error
 *
 */
struct	sockaddr_in*
__rpp_getaddr(int index)
{
	struct	stream	*sp;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, index))
	if (index < 0 || index >= stream_num) {
		errno = EINVAL;
		return NULL;
	}

	sp = &stream_array[index];
	if (sp->state <= RPP_FREE) {
		errno = ENOTCONN;
		return NULL;
	}

	return &sp->addr;
}

/**
 * @brief
 *	-return the local network address for a stream.
 *
 * @param[in] index - index of data
 *
 * @return      structure handle
 * @retval      local network address		success
 * @retval      NULL				error
 *
 */
struct	sockaddr_in*
__rpp_localaddr(int index)
{
	struct	stream	*sp;
	pbs_socklen_t	ssize = sizeof(struct sockaddr);
	static	struct sockaddr_in	lsaddr;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, index))
	if (index < 0 || index >= stream_num) {
		errno = EINVAL;
		return NULL;
	}

	sp = &stream_array[index];
	if (sp->state <= RPP_FREE) {
		errno = ENOTCONN;
		return NULL;
	}

	if (getsockname(sp->fd, (struct sockaddr *) &lsaddr, &ssize))
		return NULL;

	return &lsaddr;
}

/**
 * @brief
 *	-Free all memory and close the socket.
 *
 */
void
__rpp_terminate(void)
{
	struct	stream		*sp;
	struct	send_packet	*spp;
	struct	pending		*ppp;
	struct	recv_packet	*rpp;
	int			i;

	for (i=0; i<rpp_fd_num; i++)
#ifdef WIN32
		(void)closesocket(rpp_fd_array[i]);
#else
		(void)close(rpp_fd_array[i]);
#endif
	if (rpp_fd_array) {
		free(rpp_fd_array);
		rpp_fd_array = NULL;
		rpp_fd_num = 0;
	}
	for (i=0; i<stream_num; i++) {
		sp = &stream_array[i];

		if (sp->state == RPP_DEAD)
			continue;

		for (ppp = sp->pend_head; ppp; ppp = sp->pend_head) {
			if (ppp->data)
				free(ppp->data);
			sp->pend_head = ppp->next;
			free(ppp);
		}
		for (rpp = sp->recv_head; rpp; rpp = sp->recv_head) {
			if (rpp->data)
				free(rpp->data);
			sp->recv_head = rpp->next;
			free(rpp);
		}
		for (spp = sp->send_head; spp; spp = sp->send_head) {
			if (spp->data)
				free(spp->data);
			sp->send_head = spp->next;
			free(spp);
		}
	}
	top = NULL;
	bottom = NULL;
	if (stream_array)
		free(stream_array);
	stream_num = 0;
	stream_array = NULL;
	rpp_fd = -1;
}

/**
 * @brief
 *	-Shutdown the library.  Flush and close all open streams
 *	and call __rpp_terminate().  Should only be called by process
 *	which did the initialization.
 *
 * @see __rpp_bind
 *
 */
void
__rpp_shutdown(void)
{
	int			timeouts, num, i;
	fd_set			fdset;
	struct	timeval		tv;

	if ((pid_t)getpid() != bind_pid)
		return;	/* we are not who setup rpp so do not shut it down */

	FD_ZERO(&fdset);
	for (i=0; i<stream_num; i++)
		(void)__rpp_close(i);

	for (timeouts = 0; timeouts < 3;) {
		for (i=0; i<stream_num; i++) {
			if (stream_array[i].state > RPP_FREE)
				break;
		}
		if (i == stream_num)
			break;

		DBPRT((DBTO, "shutdown: stream %d state %d\n",
			i, stream_array[i].state))
		if ((num = __rpp_recv_all()) == -1)
			break;
		__rpp_send_out();
		if (num == -3) {        /* got nothing -- wait a bit */
			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 == 0)
				timeouts++;
			if (i == -1)
				break;
		}
	}
	__rpp_terminate();
}

/**
 * @brief
 *	-Terminate a connection stream.
 *
 * @param[in] index - index of data
 *
 * @return	int
 * @retval	0	if it all went well
 * @retval	-1	on error
 *
 */
int
__rpp_close(int index)
{
	struct	stream		*sp;

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

	switch (sp->state) {

		case RPP_STALE:
			clear_stream(sp);
			return 0;

		case RPP_CLOSE_PEND:
			sp->state = RPP_LAST_ACK;
			break;

		case RPP_OPEN_WAIT:
		case RPP_CONNECT:
			if (sp->pend_head != NULL) {
				if (__rpp_dopending(index, TRUE))
					return -1;
			}
			sp->state = RPP_CLOSE_WAIT1;
			break;

		default:
			errno = ENOTCONN;
			return -1;		/* stream closed */
	}

	__rpp_form_pkt(index, RPP_GOODBYE,
		sp->send_sequence, NULL, 0);
	if (__rpp_recv_all() == -1)
		return -1;
	__rpp_send_out();
	return 0;
}

/**
 * @brief
 *	-Add information to the stream given by index.
 *
 * @param[in] index - index of data
 * @param[in] buf - buffer with information
 * @param[in] len - length of info
 *
 * @return	int
 * @retval	-1	on error
 * @return	number of bytes written
 */
int
__rpp_write(int index, void *buf, int len)
{
	struct	stream	*sp;
	struct	pending	*pp;
	int		hold, residual, more;

	DBPRT((DBTO, "%s: entered index %d size %d\n", __func__, 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;
	errno = 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;
}

/**
 * @brief
 *	-Check a stream to see if it needs attention.
 *
 * @param[in] index - index to data
 *
 * @return	int
 * @retval	TRUE	state stale
 * @retval	msgcnt	success
 * @return	FALSE	failure
 *
 */
static
int
__rpp_attention(int index)
{
	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",
		__func__, 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;
}

/**
 * @internal
 * @brief
 *	Provide settings to control RPP.
 *
 * The value for cmd RPP_ADVISE_TIMEOUT will require ctl to point to
 * a time_t value which will be used to set read_timeout.
 *
 * @param[in]	cmd	tag for the operation being done
 * @param	ctl	generic pointer which depends on the operation
 *
 * @return	int
 *
 * @retval	0: success
 * @retval	-1: error
 */
int
rpp_advise(int cmd, void *ctl)
{
	DBPRT((DBTO, "%s: entered cmd %d\n", __func__, cmd))

	switch (cmd) {
		case RPP_ADVISE_TIMEOUT:
			if (ctl == NULL)
				break;

			read_timeout = *((time_t *)ctl);	/* save timeout value */
			DBPRT((DBTO, "%s: set read_timeout = %ld\n", __func__, read_timeout))
			return 0;
	}
	return -1;
}

/**
 * @brief
 *	-Check some state before reading or skipping.
 *
 * @param[in] index - index to data
 *
 * @return	int
 * @retval	1: okay to continue
 * @retval	0: no data
 * @retval	-1: error
 * @retval	-2: other side has closed
 */
static
int
__rpp_okay(int index)
{
	struct	stream		*sp;
	fd_set			fdset;
	struct	timeval		tv;
	time_t	start = {0};

	if (read_timeout >= 0)
		start = time(NULL);

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

		/* check to see if the read has timed out */
		if (read_timeout >= 0) {
			time_t	now = time(NULL);

			DBPRT((DBTO, "%s: waited %ld of %ld\n", __func__,
				now - start, read_timeout))
			if ((now - start) > read_timeout) {
				errno = ETIMEDOUT;
				return -1;
			}
		}

		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;
}

/**
 * @brief
 *	-Read a message.  Return data up to the end of a message
 *	or the end of the provided buffer.
 *
 * @param[in] index - index of data
 * @param[in] buf - buffer to hols data
 * @param[in] len - length of data in buf
 *
 * @return	number of bytes read
 * @retval	-1	error
 * @retval	-2	other side has closed
 */
int
__rpp_read(int index, void *buf, int len)
{
	int			hiwater, cpylen, hold, ret, xlen;
	struct	recv_packet	*pp;
	struct	stream		*sp;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, 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;
	}

	errno = 0;
	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;
}

/**
 * @brief
 *	-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.
 *
 * @param[in] indeex - index of data
 * @param[in] flag - indication to commit or not
 *
 * @return	int
 * @retval	FALSE	on decommit or if end-of-message has not been reached
 * @retval	TRUE	end-of-message has been reached
 * @retval	-1	error
 */
int
__rpp_rcommit(int	index, int flag)
{
	struct	stream		*sp;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, 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);
}

/**
 * @brief
 *	-Reset end-of-message condition on a stream.  Any packets
 *	on the receive queue are freed.
 *
 * @param[in] index - index of data
 *
 * @return	int
 * @retval	0	success
 * @retval	-1	error
 */
int
__rpp_eom(int index)
{
	struct	stream		*sp;
	struct	recv_packet	*pp;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, 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;
}

/**
 * @brief
 *	-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.
 *
 * @param[in] indeex - index of data
 * @param[in] flag - indication to commit or not
 *
 * @return	int
 * @retval	0	success
 * @retval	-1	error
 */
int
__rpp_wcommit(int index, int flag)
{
	struct	pending		*pp, *next;
	struct	stream		*sp;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, 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) {
		if (pp->data)
			free(pp->data);
		next = pp->next;
		free(pp);
	}
	sp->pend_head->next = NULL;
	sp->pend_tail = sp->pend_head;
	return 0;
}

/**
 * @brief
 *	-Skip len characters of a message.
 *
 * @param[in] index - index of data
 * @param[in] len - length of data to skip
 *
 * @return      int
 * @retval      0       success
 * @retval      -1      error
 */
int
__rpp_skip(int index, size_t len)
{
	struct	stream		*sp;
	int			ret, hiwater;

	DBPRT((DBTO, "%s: entered index %d\n", __func__, 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;
	}

	errno = 0;
	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;
}

/**
 * @brief
 *	-Check for any stream with a message waiting and
 *
 * @return	the stream number
 * @retval	-1	there are no streams with a message waiting
 */
int
__rpp_poll(void)
{
	int			i;

	DBPRT((DBTO, "%s: entered streams %d\n", __func__, stream_num))
	errno = 0;

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

/**
 * @brief
 *	-Process any stream i/o.
 *
 * @retval	0	success
 * @retval	-1	there was an error
 */
int
__rpp_io(void)
{
	int			i;

	DBPRT((DBTO, "%s: entered streams %d\n", __func__, stream_num))
	errno = 0;
	/*
	 ** 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;
}

/**
 * @brief
 *	-Read a character.
 *
 * @param[in] index -index of data
 *
 * @return	int
 * @retval	>=0	the char read
 * @retval	-1	error or EOD
 * @retval	-2	EOF
 */
int
__rpp_getc(int index)
{
	int	ret;
	u_char	c;

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

/**
 * @brief
 *	Write a character.
 *
 * @param[in] index - index of data
 * @param[in] c - character to be written
 *
 * @return	int
 * @retval	0	success
 * @retval	-1	error
 *
 */
int
__rpp_putc(int index, int c)
{
	u_char	x = (u_char)c;

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

/**
 * @brief	Destroy (delete) a stream without any i/o to close it
 *	clear_stream() is called to discard all pending packets and reset
 *	the stream structure.
 *
 * @par IMPORTANT NOTE
 * This function should only be called on an event like the daemon (server)
 * restart when it is known that the other end (mom) does not have the other
 * end of the stream open.   Unlike rpp_close() there is no attempt to send 
 * control packets back and forth for an orderly close of the stream.
 *
 * @param[in]   stream - rpp stream to distroy
 *
 * @return void
 */
void 
__rpp_destroy(int stream)
{
	struct stream  *sp;
	if (stream >= 0) {
		sp = &stream_array[stream];
		clear_stream(sp);
	}
}

/** 
 * @brief
 *	-reset the rpp setup
 *
 */
void
__DIS_rpp_reset()
{
	if (dis_getc != __rpp_getc) {
		dis_getc = __rpp_getc;
		dis_puts = (int (*)(int, const char *, size_t))__rpp_write;
		dis_gets = (int (*)(int, char *, size_t))__rpp_read;
		disr_skip   = (int (*)(int, size_t))__rpp_skip;
		disr_commit = __rpp_rcommit;
		disw_commit = __rpp_wcommit;
	}
}

/**
 * @brief
 *	rpp related functions
 */
void
set_rpp_funcs(void (*log_fn)(char *))
{
	pfn_rpp_open = __rpp_open;
	pfn_rpp_bind = __rpp_bind;
	pfn_rpp_poll = __rpp_poll;
	pfn_rpp_io   = __rpp_io;
	pfn_rpp_read = __rpp_read;
	pfn_rpp_write = __rpp_write;
	pfn_rpp_close = __rpp_close;
	pfn_rpp_destroy = __rpp_destroy;
	pfn_rpp_localaddr = __rpp_localaddr;
	pfn_rpp_getaddr = __rpp_getaddr;
	pfn_rpp_flush = __rpp_flush;
	pfn_rpp_shutdown = __rpp_shutdown;
	pfn_rpp_terminate = __rpp_terminate;
	pfn_rpp_rcommit = __rpp_rcommit;
	pfn_rpp_wcommit = __rpp_wcommit;
	pfn_rpp_skip = __rpp_skip;
	pfn_rpp_eom = __rpp_eom;
	pfn_rpp_getc = __rpp_getc;
	pfn_rpp_putc = __rpp_putc;
	pfn_DIS_rpp_reset = __DIS_rpp_reset;
	pfn_DIS_rpp_setup = __DIS_rpp_setup;
	pfn_rpp_add_close_func = NULL; /* if called with dump core, which is good */
	rpp_logfunc = log_fn;
}

