This repository has been archived on 2024-01-04. You can view files and clone it, but cannot push or open issues or pull requests.
ncsa-mosaic/libdtm/rwrtns.c

681 lines
16 KiB
C

/*****************************************************************************
*
* NCSA DTM version 2.3
* May 1, 1992
*
* NCSA DTM Version 2.3 source code and documentation are in the public
* domain. Specifically, we give to the public domain all rights for future
* licensing of the source code, all resale rights, and all publishing rights.
*
* We ask, but do not require, that the following message be included in all
* derived works:
*
* Portions developed at the National Center for Supercomputing Applications at
* the University of Illinois at Urbana-Champaign.
*
* THE UNIVERSITY OF ILLINOIS GIVES NO WARRANTY, EXPRESSED OR IMPLIED, FOR THE
* SOFTWARE AND/OR DOCUMENTATION PROVIDED, INCLUDING, WITHOUT LIMITATION,
* WARRANTY OF MERCHANTABILITY AND WARRANTY OF FITNESS FOR A PARTICULAR PURPOSE
*
*****************************************************************************/
/***************************************************************************
**
** rwrtns.c - provides very low level routines for reading and writing
** buffer. This code should be independent of communication
** channel as long as the descriptor will work with system
** read and write routines.
**
***************************************************************************/
/*********************************************************************
**
** $Header: /X11/mosaic/cvsroot/xmosaic3/libdtm/rwrtns.c,v 1.5 1996/06/06 19:48:08 spowers Exp $
**
**********************************************************************/
/*
#ifdef RCSLOG
$Log: rwrtns.c,v $
Revision 1.5 1996/06/06 19:48:08 spowers
Linux is brain ded.
Revision 1.4 1996/02/18 23:40:16 spowers
PROTO -> DTM_PROTO
Revision 1.3 1995/11/10 12:03:32 spowers
dumb warning messages
Revision 1.2 1995/10/13 06:33:23 spowers
Solaris support added.
Revision 1.1.1.1 1995/01/11 00:03:02 alanb
New CVS source tree, Mosaic 2.5 beta 4
* Revision 2.5 1994/12/29 23:40:14 alanb
* I'm committing with a new symbolic revision number.
*
* Revision 1.1.1.1 1994/12/28 21:37:33 alanb
*
* Revision 1.2 1993/10/29 03:46:49 marca
* Tweaks.
*
* Revision 1.1.1.1 1993/07/04 00:03:13 marca
* Mosaic for X version 2 distribution
*
* Revision 1.1 1993/01/18 21:50:37 marca
* I think I got it now.
*
* Revision 1.19 92/05/14 19:27:48 jefft
* modified dtm_recv_reliable
*
* Revision 1.18 1992/04/30 20:25:27 jplevyak
* Changed Version to 2.3.
*
* Revision 1.17 1992/04/29 22:01:34 jplevyak
* Fix big with exact size buffers. Remove dead code.
*
* Revision 1.16 1992/04/13 16:07:10 jplevyak
* Changes for DEC and RS6000.
*
* Revision 1.15 92/03/10 22:07:10 jplevyak
* Added changed for PC/MAC from Quincey Koziol (koziol@ncsa.uiuc.edu)
* with modification.
*
* Revision 1.14 1992/03/02 18:29:41 jplevyak
* Fixed bug in EAGAIN handling.
*
* Revision 1.13 1992/03/02 17:20:14 jplevyak
* Temporary back out.
*
* Revision 1.11 1992/02/27 23:44:31 jplevyak
* Surrounded writes by code intended to recover from signal interruptions.
*
* Revision 1.10 1992/01/14 16:31:40 creiman
* Removed mac #include
*
* Revision 1.9 1991/10/11 20:26:23 jplevyak
* Fixed incorrect #def use.
*
* Revision 1.8 1991/10/10 15:14:01 jplevyak
* Fixed naming convensions.
*
* Revision 1.7 91/08/20 15:56:06 sreedhar
* Removed unused functions - dtm_write_buffer, dtm_send, dtm_recv
*
* Revision 1.6 1991/08/15 18:56:52 sreedhar
* Changes for logical portname version
*
* Revision 1.4 1991/06/11 15:19:51 sreedhar
* disclaimer added
*
* Revision 1.3 1991/06/07 16:06:29 sreedhar
* sizeof( int ) replaced by 4 for message to be sent out
*
* Revision 1.2 1991/05/30 15:51:50 sreedhar
* Changes for readMsg/writeMsg internal release
*
* Revision 1.1 1990/11/08 16:38:13 jefft
* Initial revision
*
#endif
*/
#include <stdio.h>
#include <sys/types.h>
#if defined(_ARCH_MSDOS)
#include <nmpcip.h>
#include <time.h>
#include "uio.h"
#else
#include <sys/socket.h>
#include <sys/ioctl.h>
#ifdef LINUX
#include "linux-uio.h"
#else
#include <sys/uio.h>
#endif
#include <netinet/in.h>
#endif
#include <fcntl.h>
#include <errno.h>
#include <sys/time.h>
#ifdef SOLARIS
#include <sys/filio.h>
#endif
#ifdef RS6000
#include <sys/select.h>
#endif
#include "dtmint.h"
#include "debug.h"
/*
CONTENTS
dtm_read_buffer() - attempts to fill the next dtm buffer.
dtm_recv_header() - Function to read header and return size.
dtm_recv_ack() - receive message ackowledgement
tm_send_ack() - send message acknowledgement
dtm_writev_buffer() - sends the buffers to receiving process.
*/
/*
STATIC FUNCTION PROTOTYPES
*/
#ifdef DTM_PROTOTYPES
static int dtm_recv_reliable DTM_PROTO((int ,char *,int ));
static int dtm_writev_failed DTM_PROTO((int ,struct msghdr *,int ));
static int dtm_send_some DTM_PROTO((int d, char *buf, int bufsize ));
#endif
static int padding[] = {0, 3, 2, 1};
/* Technique from XlibInt.c
*/
#if defined(EAGAIN) && defined(EWOULDBLOCK)
#define ERRTEST(err) (err == EAGAIN || err == EWOULDBLOCK)
#else
#if defined(EAGAIN)
#define ERRTEST(err) (err == EAGAIN)
#else
#define ERRTEST(err) (err == EWOULDBLOCK)
#endif
#endif
#ifdef DTM_PROTOTYPES
static int ready_bytes(int d, int length )
#else
static int ready_bytes( d, length )
int d, length;
#endif
{
int num;
fd_set mask;
struct timeval timeout ;
/* set the select timeout value */
timeout.tv_sec = 2;
timeout.tv_usec = 0;
FD_ZERO(&mask);
FD_SET(d, &mask);
#ifdef __hpux
num = select(FD_SETSIZE, (int *)&mask, (int *)0, (int *)0, &timeout);
#else
num = select(FD_SETSIZE, &mask, (fd_set *)0, (fd_set *)0, &timeout);
#endif
if (num < 0) {
DTMerrno = DTMSELECT;
return DTMERROR;
}
else if (num == 0) {
DTMerrno = DTMTIMEOUT;
return DTMERROR;
}
else {
ioctl(d, FIONREAD, &num);
if (num < length) {
DTMerrno = DTMTIMEOUT;
return DTMERROR;
}
else
return DTM_OK;
}
}
/*
Reliably read from a port in the face of signals and other
'errors' produced by the operating system.
*/
#ifdef DTM_PROTOTYPES
static int dtm_recv_reliable(int d,char *buffer,int length )
#else
int dtm_recv_reliable( d, buffer, length )
int d;
char * buffer;
int length;
#endif
{
int bytes_read;
while ( (bytes_read = recv( d, buffer, length, 0)) != length ) {
if ( bytes_read > 0) {
length -= bytes_read;
buffer += bytes_read;
} else if (ERRTEST(errno)) {
fd_set filedes;
int got;
/* FD_ZERO and FD_SET were moved into the select loop */
/* just in case the select is clearing filedes */
do {
FD_ZERO( &filedes );
FD_SET( d, &filedes );
#ifdef __hpux
got = select( d, (int *)&filedes, (int *)NULL, (int *)NULL,
#else
got = select( d, &filedes, (fd_set *)NULL, (fd_set *)NULL,
#endif
NULL );
if (got < 0 && errno != EINTR ) {
DTMerrno = DTMREAD;
return DTMERROR;
}
} while ( got <= 0 );
continue;
} else if (bytes_read == 0) {
DTMerrno = DTMEOF;
return DTMERROR;
} else if (errno != EINTR) {
DTMerrno = DTMREAD;
return DTMERROR;
}
}
return DTM_OK;
}
/*
* dtm_read_buffer() - attempts to fill the next dtm buffer. The
* blocklen variable must be set to DTM_NEW_DATASET after each dataset
* to force recv_buffer to move the next dataset.
*/
#ifdef DTM_PROTOTYPES
int dtm_read_buffer(int d,int32 *blocklen,VOIDPTR buffer,int length)
#else
int dtm_read_buffer(d, blocklen, buffer, length)
int d, *blocklen;
VOIDPTR buffer;
int length;
#endif
{
reg int tmp, readcnt, count = 0;
DBGFLOW("# dtm_read_buffer called.\n");
DBGMSG1("dtm_recv_buffer: attempting to read %d bytes.\n", length);
DBGMSG1("dtm_recv_buffer: initial blocklen = %d\n", *blocklen);
/* if block length is DTM_NEW_DATASET this is a new dataset
* get initial block count
*/
if (*blocklen == DTM_NEW_DATASET) {
CHECK_ERR(dtm_recv_reliable(d, (char *)blocklen, 4));
LOCALINT(*blocklen);
DBGINT("initial blocklen = %d\n", *blocklen);
}
/* attempt to get a full buffer */
while (TRUE) {
/* if block length is 0, because last call to fill_buffer hit
* the EOS or because this dataset is zero length, return 0
* to indicate the end of dataset.
*/
if (*blocklen == 0)
return 0;
/* if block length is greater than buffer size then... */
if (*blocklen >= length - count) {
readcnt = length - count;
CHECK_ERR( dtm_recv_reliable( d, ((char *)buffer) + length - readcnt,
readcnt));
/* decrement block length, if 0 get next block length */
*blocklen -= (length - count);
if (*blocklen == 0)
*blocklen = DTM_NEW_DATASET;
/* if block length is 0 now, the EOS will be returned on */
/* the next call to fill_buffer */
/* return full buffer count */
DBGINT("recv_buffer: buffer full, returning %d\n", length);
return length;
}
/* else block length is less than buffer size */
else {
readcnt = *blocklen;
CHECK_ERR( dtm_recv_reliable( d, (char *)buffer + count +
*blocklen - readcnt, readcnt));
/* increment count */
count += *blocklen;
/* get next block length */
CHECK_ERR( dtm_recv_reliable(d, (char *)blocklen, 4));
LOCALINT(*blocklen);
DBGINT("blocklen = %d\n", *blocklen);
/* if block length is 0 now, the correct count will be */
/* returned now, and EOS on the next call to fill_buffer */
if (*blocklen == 0)
return count;
}
} /* end while */
}
/*
Replaces dtm_recv_header for nornal communication.
*/
#ifdef DTM_PROTOTYPES
int dtm_read_header(int fd,void *buf,int buflen )
#else
int dtm_read_header( fd, buf, buflen )
int fd;
void * buf;
int buflen;
#endif
{
int32 hdrsize;
CHECK_ERR(ready_bytes(fd, 4));
CHECK_ERR( dtm_recv_reliable( fd, (char *)&hdrsize, 4 ));
LOCALINT(hdrsize);
if ( hdrsize <= buflen ) {
CHECK_ERR( dtm_recv_reliable( fd, buf, hdrsize ));
return hdrsize;
} else {
CHECK_ERR( dtm_recv_reliable( fd, buf, buflen ));
{
int left = hdrsize - buflen;
int readcnt = left % DISCARDSIZE;
if (!readcnt) readcnt = DISCARDSIZE;
while (left) {
CHECK_ERR(dtm_recv_reliable( fd, dtm_discard, readcnt ));
left -= readcnt;
readcnt = DISCARDSIZE;
}
}
}
DTMerrno = DTMHEADER;
return DTMERROR;
}
/*
dtm_recv_header()
Function to read header and return size.
Notes : If buffer is too small, dump remainder of header
and return error.
Actually, this is function to read length of data and
then to receive that much data - the data is called header
everywhere since that was the first usage of the function.
*/
#ifdef DTM_PROTOTYPES
int dtm_recv_header(int d,VOIDPTR header,int length )
#else
int dtm_recv_header( d, header, length )
int d;
int length;
VOIDPTR header;
#endif
{
int readcnt, headerlen, tmp;
struct sockaddr_in from ;
int fromlen = sizeof( struct sockaddr_in ) ;
DBGFLOW("# dtm_recv_header called.\n");
DBGMSG1("dtm_recv_header: fd = %d.\n", d);
DBGMSG1("dtm_recv_header: buf length = %d.\n", length);
/* get header length */
if( (readcnt = recvfrom(d, (char *)&headerlen, 4, 0, ( struct sockaddr *)&from,
( int *)&fromlen)) != 4) {
/* somehow hit EOF, return DTMEOF instead */
if( readcnt == 0 ) {
DTMerrno = DTMEOF;
DBGMSG("dtm_recv_header: EOF1.\n");
return DTMERROR;
} else {
if( errno == ECONNRESET ) {
/* connection closed by writer, return EOF */
DBGMSG("dtm_recv_header: EOF2.\n");
DTMerrno = DTMEOF;
return DTMERROR;
} else {
/* don't know what the problem is, punt... */
DBGMSG("dtm_recv_header: EOF3.\n");
DTMerrno = DTMREAD;
return DTMERROR;
}
}
}
LOCALINT(headerlen);
DBGMSG("dtm_recv_header: got length.\n");
/* read the header */
readcnt = (length > headerlen) ? headerlen : length ;
header = (void *) (((char *) header) + readcnt);
while(readcnt) {
if( (tmp = recvfrom(d, ((char *)header) - readcnt, readcnt, 0,
( struct sockaddr *)&from, ( int *)&fromlen)) > 0)
readcnt -= tmp;
else {
DTMerrno = DTMREAD;
return DTMERROR;
}
}
/* check for header greater than buffer size provided */
if( length >= headerlen )
return headerlen;
else {
/* discard remaining header */
readcnt = headerlen - length;
while (readcnt) {
if ((tmp = recvfrom(d, dtm_discard, readcnt, 0,
(struct sockaddr *)&from, (int *)&fromlen)) > 0)
readcnt -= tmp;
else {
DTMerrno = DTMREAD;
return DTMERROR;
}
}
DTMerrno = DTMHEADER;
return DTMERROR;
}
}
/*
dtm_recv_ack() - receive message ackowledgement
Notes : Berkeley implementation returns 0 from recv
if socket connection breaks while waiting in
recv system call. System V returns -1 and
ECONNRESET in errno for same error.
For historical reasons, DTMEOF is returned when
socket connection breaks in middle instead of
say DTMFCONN ( DTM connection failed error )
*/
#ifdef DTM_PROTOTYPES
int dtm_recv_ack(int d,int *ack )
#else
int dtm_recv_ack( d, ack )
int d;
int *ack;
#endif
{
int tmp ;
DBGFLOW("# dtm_recv_ack called.\n");
/* there should be no possibility of blocking after this call */
CHECK_ERR(ready_bytes(d, 4));
if( (tmp = recv( d, (char *)ack, 4, 0 )) != 4 ) {
DBGINT( "Recv_ack errno = %d\n", errno ) ;
if( tmp == 0 )
/* Courtesy Berkeley */
DTMerrno = DTMEOF ;
else {
if( errno == ECONNRESET )
/* Courtesy system V */
DTMerrno = DTMEOF;
else
DTMerrno = DTMREAD;
}
return DTMERROR;
}
DBGMSG1( "ack received, tmp = %d\n", tmp );
LOCALINT(*ack);
return DTM_OK;
}
#ifdef DTM_PROTOTYPES
static int dtm_send_some(int d, char *buf, int bufsize )
#else
int dtm_send_some( d, buf, bufsize )
int d;
char * buf;
int bufsize;
#endif
{
int tmp ;
while (bufsize ) {
tmp = send(d, buf, bufsize, 0);
if ( tmp >= 0 ) {
bufsize -= tmp;
buf += tmp;
continue;
}
if (errno == EPIPE) {
/* socket connection broke in middle */
DTMerrno = DTMEOF ;
return DTMERROR;
} else if ( ERRTEST( errno ) ) {
fd_set filedes;
int got;
FD_ZERO( &filedes );
FD_SET( d, &filedes );
do {
#ifdef __hpux
got = select( 32, (int *)&filedes, (int *)NULL, (int *)NULL,
#else
got = select( 32, &filedes, (fd_set *)NULL, (fd_set *)NULL,
#endif
NULL );
if (got < 0 && errno != EINTR ) {
DTMerrno = DTMWRITE;
return DTMERROR;
}
} while ( got <= 0 );
continue;
} else DTMerrno = DTMWRITE ;
return DTMERROR;
}
return DTM_OK;
}
/*
* dtm_send_ack() - send message acknowledgement
*/
#ifdef DTM_PROTOTYPES
int dtm_send_ack(int d, int32 ack)
#else
int dtm_send_ack(d, ack)
int d;
int32 ack;
#endif
{
DBGFLOW("# dtm_send_ack called.\n");
STDINT(ack);
return dtm_send_some( d, (char *)&ack, 4 );
}
#ifdef DTM_PROTOTYPES
static int dtm_writev_failed(int fd,struct msghdr *msgbuf,int tmp )
#else
int dtm_writev_failed( fd, msgbuf, tmp )
int fd;
struct msghdr * msgbuf;
int tmp;
#endif
{
int done = tmp;
int i;
struct iovec * iov;
iov=msgbuf->msg_iov;
if ( tmp < 0 ) done = 0;
for ( i = 0; i < msgbuf->msg_iovlen; i++ ) {
done -= iov[i].iov_len;
if ( done > 0 ) continue;
if ( dtm_send_some( fd, iov[i].iov_base + done + iov[i].iov_len,
(- done )) == DTMERROR )
return DTMERROR;
done = 0;
}
}
/*
dtm_writev_buffer() - sends the buffers to receiving process.
*/
#ifdef DTM_PROTOTYPES
int dtm_writev_buffer(int fd,struct iovec *iov,int32 iovlen,int32 iovsize,
struct sockaddr *addr,int addrlen )
#else
int dtm_writev_buffer( fd, iov, iovlen, iovsize, addr, addrlen )
int fd ;
struct iovec *iov ;
int32 iovlen ;
int32 iovsize ;
struct sockaddr *addr ;
int addrlen ;
#endif
{
int tmp;
struct msghdr msgbuf ;
int todo;
DBGINT("# dtm_writev_buffer called, fd %d.\n", fd );
msgbuf.msg_name = (void *)addr ;
msgbuf.msg_namelen = addrlen ;
msgbuf.msg_iov = iov ;
msgbuf.msg_iovlen = iovlen ;
msgbuf.msg_accrights = 0 ;
if( (tmp = sendmsg( fd, &msgbuf, 0 )) != iovsize )
return dtm_writev_failed( fd, &msgbuf, tmp );
DBGINT( "dtm_writev_buffer tmp = %d\n", tmp );
return DTM_OK ;
}