sockrw.c

Path MPICH-V for MPICH the MPI Implementation

/** @file sockrw.c implements the basic socket methods  */
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <netinet/in.h>
#include <netinet/tcp.h>
#include <netdb.h>
#include <unistd.h>
#include <fcntl.h>
#include "debug.h"

#include "utils_socket.h"

/*
 * Create a new socket with some special options.
 * @return a new socket (TCP)
 */
int _usocket(void)
{
  int sock, optval;

  sock = socket(AF_INET, SOCK_STREAM, 0);
  if (sock < 0) qerror("ERROR opening socket");

  /* Remove the delay for small packets */
  optval = 1;
  if( setsockopt(sock, IPPROTO_TCP, TCP_NODELAY, (char *) &optval, sizeof(optval)) == -1 )
    qerror("setsockopt: %s\n",strerror(errno));
  /* Set the size of the reception buffer */
  optval = MAX_BUF_FOR_SOCKET;
  if( setsockopt(sock, SOL_SOCKET, SO_RCVBUF, (char *) &optval, sizeof(optval)) == -1 )
    qerror("setsockopt: %s\n",strerror(errno));
  /* and set the size of the send buffer */
  optval = MAX_BUF_FOR_SOCKET;
  if( setsockopt(sock, SOL_SOCKET, SO_SNDBUF, (char *) &optval, sizeof(optval)) == -1 )
    qerror("setsockopt: %s\n",strerror(errno));
  return sock;
}

/*
 * @param sock the socket fd from where to read something
 * @param buf a pojter to the buffer where the message have to be stored
 * @param size the size of the buffer
 *
 * @return number of bytes already reads
 */
int _urecv(int sock, void *buf, int size, int flag )
{
  int n, rd = 0;
  
  while(rd < size)
  {
    n = recv(sock, buf + rd, size - rd, MSG_WAITALL | flag);
    if(n == 0)
    {
      /** change mode to synchronous, since this 0 may be not disconnect */
      int fdflags = fcntl(sock, F_GETFL);
      int nflags  = (fdflags & (~(O_ASYNC|O_NONBLOCK))) | (O_SYNC);
    again:
      fcntl(sock, F_SETFL, nflags);
      n = recv(sock, buf + rd, size - rd, MSG_WAITALL | flag);
      /** back to old mode */
      fcntl(sock, F_SETFL, fdflags);
      if(n == 0)
	{
	  /** this is a disconnection (sure) */
	  return 0;
	}
      if(n < 0)
	{
	  if(errno == EINTR) 
	    goto again;
	  printe("SyncReading %d/%d bytes on fd %d before error", rd, size, sock);
	  return -1;
	}
    }
    if(n < 0)
    {
      if((errno == EINTR)|| (errno == EAGAIN)) continue;
      printe("SyncReading %d/%d bytes on fd %d before error", rd, size, sock);
      return -1;
    }
    rd += n;
  }
  return rd;
}

int _uread(int fd, void *buf, int size)
{
  int n, rd = 0;
  
  while(rd < size)
    {
      n = read(fd, buf + rd, size - rd);
      if(n <= 0)
	{
	  printi("read", "SyncReading %d/%d bytes on fd %d before error : %d (%s)",
			  rd, size, fd, errno, strerror(errno));
	  if((errno == EINTR) || (errno == EAGAIN)) 
	    continue;
	  printe("SyncReading %d/%d bytes on fd %d before error", rd ,size, fd);
	  return -1;
	}
      rd += n;
    }
  return rd;
}

/*
 * @param sock the socket fd where to write something
 * @param buf the buffer containing the data to be send
 * @param size the size of the buffer
 *
 * @return number of bytes already written
 */
int _usend(int sock, void *buf, int size, int flag)
{
    int n, written = 0;

    while( written < size ) {
	n = send( sock, buf+written, size-written, MSG_NOSIGNAL | flag );
	if( n < 0 ) {
	    if((errno == EINTR) || (errno == EAGAIN)) continue;
#ifdef DEBUG
	    printe( "Writing %d/%d bytes on %d  before error",
		       written, size, sock);
#endif
	    return n;
	}
	written += n;
    }
    return written;
}

int _uwrite(int fd, void *buf, int size)
{
    int n, written = 0;

    while( written < size ) 
      {
	n = write(fd, buf+written, size-written);
	if( n < 0 ) 
	  {
	    if((errno == EINTR) || (errno == EAGAIN)) 
	      continue;
	    printe( "Writing %d/%d bytes on %d  before error",
		    written, size, fd);
	    return n;
	  }
	written += n;
      }
    return written;
}