rpp.c

OpenPBS

/*
*         OpenPBS (Portable Batch System) v2.3 Software License
* 
* Copyright (c) 1999-2002 Veridian Information Solutions, Inc.
* All rights reserved.
* 
* ---------------------------------------------------------------------------
* For a license to use or redistribute the OpenPBS software under conditions
* other than those described below, or to purchase support for this software,
* please contact Veridian Systems, PBS Products Department ("Licensor") at:
* 
*    www.OpenPBS.org  [phone]                  [e-mail]
*                        877 902-4PBS (US toll-free)
* ---------------------------------------------------------------------------
* 
* This license covers use of the OpenPBS v2.3 software (the "Software") at
* your site or location, and, for certain users, redistribution of the
* Software to other sites and locations.  Use and redistribution of
* OpenPBS v2.3 in source and binary forms, with or without modification,
* are permitted provided that all of the following conditions are met.
* After December 31, 2003, only conditions 3-6 must be met:
* 
* 1. Commercial and/or non-commercial use of the Software is permitted
*    provided a current software registration is on file at www.OpenPBS.org.
*    If use of this software contributes to a publication, product, or
*    service, proper attribution must be given; see www.OpenPBS.org/credit.html
* 
* 2. Redistribution in any form is only permitted for non-commercial,
*    non-profit purposes.  There can be no charge for the Software or any
*    software incorporating the Software.  Further, there can be no
*    expectation of revenue generated as a consequence of redistributing
*    the Software.
* 
* 3. Any Redistribution of source code must retain the above copyright notice
*    and the acknowledgment contained in paragraph 6, this list of conditions
*    and the disclaimer contained in paragraph 7.
* 
* 4. Any Redistribution in binary form must reproduce the above copyright
*    notice and the acknowledgment contained in paragraph 6, this list of
*    conditions and the disclaimer contained in paragraph 7 in the
*    documentation and/or other materials provided with the distribution.
* 
* 5. Redistributions in any form must be accompanied by information on how to
*    obtain complete source code for the OpenPBS software and any
*    modifications and/or additions to the OpenPBS software.  The source code
*    must either be included in the distribution or be available for no more
*    than the cost of distribution plus a nominal fee, and all modifications
*    and additions to the Software must be freely redistributable by any party
*    (including Licensor) without restriction.
* 
* 6. All advertising materials mentioning features or use of the Software must
*    display the following acknowledgment:
* 
*     "This product includes software developed by NASA Ames Research Center,
*     Lawrence Livermore National Laboratory, and Veridian Information 
*     Solutions, Inc.
*     Visit www.OpenPBS.org for OpenPBS software support,
*     products, and information."
* 
* 7. DISCLAIMER OF WARRANTY
* 
* THIS SOFTWARE IS PROVIDED "AS IS" WITHOUT WARRANTY OF ANY KIND. ANY EXPRESS
* OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
* OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, AND NON-INFRINGEMENT
* ARE EXPRESSLY DISCLAIMED.
* 
* IN NO EVENT SHALL VERIDIAN CORPORATION, ITS AFFILIATED COMPANIES, OR THE
* U.S. GOVERNMENT OR ANY OF ITS AGENCIES BE LIABLE FOR ANY DIRECT OR 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.
* 
* This license will be governed by the laws of the Commonwealth of Virginia,
* without reference to its choice of law rules.
*/
/*
**	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.
*/
static	char	ident[] = "@(#) $RCSfile: rpp.c,v $ $Revision: 2.3.2.2.2.7 $";

#if !defined(_BSD) && defined(_AIX)   /* this is needed by AIX */
#define	_BSD	1
#endif

#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	<netinet/in.h>
#include	"rpp.h"

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

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

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

/*
**	Size-Constants for the various parts of RPP packet
*/
#define	RPP_PKT_SIZE	4*1024 /* max packet length, including header */
#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
/*
**	Default number of sendto attempts on a *packet.
*/
#define	RPP_RETRY	10
/*
**	Max allowed number of outstanding pkts
*/
#define	RPP_HIGHWATER	24


/*
**	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	sent_out;            /* number of times sent; <= sp->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 */

	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			retry;		/* sendto retry limit */

	int			open_key;	/* unique bit pattern created */
						/* by the end issuing the */
						/* rpp_open.  It's the same */
						/* 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 connect */
	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 date is linked */
						/* between top and bottom. */

/*
**	Global Variables
*/
int		rpp_dbprt = 0;			/* controls debug printing */

/*
**	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 set into each new stream.
*/
int		rpp_retry = RPP_RETRY;

/*
**	Tables used by the macros I2TOH, HTOI2, I8TOH, HTOI8
**	to convert 2 and 8 digit hexidecimal strings to integer and back.
*/
char	cval[] = {
	-1, -1, -1, -1, -1, -1, -1, -1,		/* nul .. bel */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* bs .. si */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* dle .. etb */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* can .. us */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* sp .. ' */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* ( .. / */
	 0,  1,  2,  3,  4,  5,  6,  7,		/* 0 .. 7 */
	 8,  9, -1, -1, -1, -1, -1, -1,		/* 8 .. ? */
	-1, 10, 11, 12, 13, 14, 15, -1,		/* @ .. G */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* H .. O */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* P .. W */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* X .. _ */
	-1, 10, 11, 12, 13, 14, 15, -1,		/* ` .. g */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* h .. o */
	-1, -1, -1, -1, -1, -1, -1, -1,		/* p .. w */
	-1, -1, -1, -1, -1, -1, -1, -1		/* 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.
*/