p4_sock_util.c

MPICH是MPI的重要研究,提供了一系列的接口函数,为并行计算的实现提供了编程环境.

    if (rc < 0)
	p4_error("net_setup_anon_listener listen", -1);

    getsockname(*skt, (struct sockaddr *) & s_in, &sinlen);
    *port = ntohs(s_in.sin_port);
}

/*
  Accept a connection on socket skt and return fd of new connection.
  */
int net_accept(int skt)
{
    struct sockaddr_in from;
    int rc, flags, skt2, gotit, sockbuffsize;
    p4_sockopt_len_t fromlen;
    int optval = P4_TRUE;

    /* dump_sockinfo("net_accept call of dumpsockinfo \n",skt); */
    fromlen = sizeof(from);
    gotit = 0;
    while (!gotit)
    {
	p4_dprintfl(60, "net_accept - waiting for accept on %d.\n",skt);
	SYSCALL_P4(skt2, accept(skt, (struct sockaddr *) &from, &fromlen));
	if (skt2 < 0)
	    p4_error("net_accept accept", skt2);
	else
	    gotit = 1;
	p4_dprintfl(60, "net_accept - got accept\n");
    }

#if defined(CAN_DO_SETSOCKOPT) && !defined(SET_SOCK_BUF_SIZE)
    net_set_sockbuf_size(-1,skt2);     /* 7/12/95, [e-mail] */
#endif

#ifdef CAN_DO_SETSOCKOPT
    SYSCALL_P4(rc, setsockopt(skt2, IPPROTO_TCP, TCP_NODELAY, (char *) &optval, sizeof(optval)));

    sockbuffsize = p4_default_sock_buf_size;

#ifdef SET_SOCK_BUF_SIZE
      if (setsockopt(skt2,SOL_SOCKET,SO_RCVBUF,(char *)&sockbuffsize,sizeof(sockbuffsize)))
      p4_dprintf("net_accept: setsockopt rcvbuf failed\n");
      if (setsockopt(skt2,SOL_SOCKET,SO_SNDBUF,(char *)&sockbuffsize,sizeof(sockbuffsize)))
      p4_dprintf("net_accept: setsockopt sndbuf failed\n");
#endif

    if (p4_debug_level > 79)
	p4_print_sock_params( skt2 );
#endif
    /* Peter Krauss suggested eliminating these lines for HPs  */
    flags = fcntl(skt2, F_GETFL, 0);
    if (flags < 0)
	p4_error("net_accept fcntl1", flags);
#   if defined(HP)
    flags |= O_NONBLOCK;
#   else
    flags |= O_NDELAY;
#   endif
#   if defined(RS6000)
    flags |= O_NONBLOCK;
#   endif
    flags = fcntl(skt2, F_SETFL, flags);
    if (flags < 0)
	p4_error("net_accept fcntl2", flags);
    return (skt2);
}

static void 
get_sock_info_by_hostname(char *hostname, struct sockaddr_in **sockinfo)
{
#ifndef P4_WITH_MPD
    int i;

    p4_dprintfl( 91, "Starting get_sock_info_by_hostname\n");
    if (p4_global) {
	p4_dprintfl( 90, "looking at %d hosts\n", 
		    p4_global->num_in_proctable );
	for (i = 0; i < p4_global->num_in_proctable; i++) {
	    p4_dprintfl(90,"looking up (%s), looking at (%s)\n",
			hostname,p4_global->proctable[i].host_name);
	    if (strcmp(p4_global->proctable[i].host_name,hostname) == 0) {
#ifdef LAZY_GETHOSTBYNAME
	      p4_procgroup_setsockaddr( &p4_global->proctable[i] );
#endif
		if (p4_global->proctable[i].sockaddr.sin_port == 0)
		    p4_error( "Uninitialized sockaddr port",i);
		*sockinfo = &(p4_global->proctable[i].sockaddr);
		return;
		}
	    }
	}
#endif

/* Error, no sockinfo.
   Try to get it from the hostname (this is NOT signal safe, so we 
   had better not be in a signal handler.  This MAY be ok for the listener) */
    p4_dprintfl(40, "get_sock_info_by_hostname: calling gethostbyname for %s\n",
      hostname);
    {
    struct hostent *hp = gethostbyname_p4( hostname );
    static struct sockaddr_in listener_sockaddr;
    if (hp) {
	bzero((P4VOID *) &listener_sockaddr, sizeof(listener_sockaddr));
	if (hp->h_length != 4)
	    p4_error("get_sock_info_by_hostname: hp length", hp->h_length);
	bcopy((P4VOID *) hp->h_addr, (P4VOID *) &listener_sockaddr.sin_addr, 
	      hp->h_length);
	listener_sockaddr.sin_family = hp->h_addrtype;
	*sockinfo = &listener_sockaddr;
	return;
	}
    }

    *sockinfo = 0;
    p4_error("Unknown host in getting sockinfo from proctable",-1);
}

/*
 * We must be careful here is using the sockinfo information from 
 * get_sock_info_by_hostname.  That routine returns a *pointer* to the
 * socket info, which is ok for readonly data, but we will need to 
 * have a modifiable version (so that we can set the indicated port).  
 * Thus, we first get a pointer to the readonly structure, then make 
 * a local copy of it.  Thanks to Peter Wycoff for finding this.
 */
int net_conn_to_listener(char *hostname, int port, int num_tries)
{
    int flags, rc, s;
    struct sockaddr_in sockinfo;
    struct sockaddr_in *sockinfo_ro; /* _ro for Read-Only */
    P4BOOL optval = P4_TRUE;
    P4BOOL connected = P4_FALSE;

    p4_dprintfl(80, "net_conn_to_listener: host=%s port=%d\n", hostname, port);
    /* gethostchange -RL */
/*
    bzero((P4VOID *) &listener, sizeof(listener));
    bcopy((P4VOID *) hp->h_addr, (P4VOID *) &listener.sin_addr, hp->h_length);
    listener.sin_family = hp->h_addrtype;
    listener.sin_port = htons(port);
*/
    get_sock_info_by_hostname(hostname,&sockinfo_ro);
    memcpy(&sockinfo, sockinfo_ro, sizeof(sockinfo));
    sockinfo.sin_port = htons(port);
#if !defined(CRAY)
    dump_sockaddr("sockinfo",&sockinfo);
#endif
    connected = P4_FALSE;
    s = -1;
    while (!connected && num_tries)
    {
	SYSCALL_P4(s, socket(AF_INET, SOCK_STREAM, 0));
	if (s < 0)
	    p4_error("net_conn_to_listener socket", s);

	p4_dprintfl(80,"net_conn_to_listener socket fd=%d\n", s );
#ifdef CAN_DO_SETSOCKOPT
        net_set_sockbuf_size(-1,s);    /* 7/12/95, [e-mail] */
	SYSCALL_P4(rc, setsockopt(s,IPPROTO_TCP,TCP_NODELAY,(char *) &optval,sizeof(optval)));
	if (p4_debug_level > 79)
	    p4_print_sock_params( s );
#       endif

	SYSCALL_P4(rc, connect(s, (struct sockaddr *) &sockinfo,
			       sizeof(struct sockaddr_in)));
	if (rc < 0)
	{
	    /* Since the socket is not yet non-blocking, EINPROGRESS should not
	       happen.  Other errors are fatal to the socket */
	    p4_dprintfl( 70, "Connect failed; closed socket %d\n", s );
	    if (p4_debug_level > 70) {
		/* Give the reason that the connection failed. */
		perror("Connection failed for reason: ");
	    }
	    close(s);
	    s = -1;
	    if (--num_tries)
	    {
		p4_dprintfl(60,"net_conn_to_listener: connect to %s failed; will try %d more times \n",hostname,num_tries);
		sleep(2);
	    }
	}
	else
	{
	    connected = P4_TRUE;
	    p4_dprintfl(70,"net_conn_to_listener: connected to %s\n",hostname);
	}
    }
    if (!connected)
	return(-1);

    /* Peter Krauss suggested eliminating these lines for HPs */
    flags = fcntl(s, F_GETFL, 0);
    if (flags < 0)
	p4_error("net_conn_to_listener fcntl1", flags);
#   if defined(HP)
    flags |= O_NONBLOCK;
#   else
    flags |= O_NDELAY;
#   endif
#   if defined(RS6000)
    flags |= O_NONBLOCK;
#   endif
    flags = fcntl(s, F_SETFL, flags);
    if (flags < 0)
	p4_error("net_conn_to_listener fcntl2", flags);

    return (s);
}

int net_recv(int fd, P4VOID *in_buf, int size)
{
    int recvd = 0;
    int n;
    int read_counter = 0;
    int block_counter = 0;
    int eof_counter = 0;
    char *buf = (char *)in_buf;
    int set_fd_blocking = 0;
    int orig_flags = 0;  /* Set to keep gcc quiet */
#if defined(P4SYSV) && !defined(NONBLOCKING_READ_WORKS)
    int n1 = 0;
    struct timeval tv;
    fd_set read_fds;
    int rc;
    char tempbuf[1];
#endif
    COLLECT_STAT(int n_loop = 0;);

    COLLECT_STAT(n_recv_calls++);

    p4_dprintfl( 99, "Beginning net_recv of %d on fd %d\n", size, fd );
    while (recvd < size)
    {
	read_counter++;

	SYSCALL_P4(n, read(fd, buf + recvd, size - recvd));
	if (n == 0)		/* maybe EOF, maybe not */
#if defined(P4SYSV) && !defined(NONBLOCKING_READ_WORKS)
	{
	    eof_counter++;

	    tv.tv_sec = 5;
	    tv.tv_usec = 0;
	    FD_ZERO(&read_fds);
	    FD_SET(fd, &read_fds);
	    SYSCALL_P4(n1, select(fd+1, &read_fds, 0, 0, &tv));
	    if (n1 == 1  &&  FD_ISSET(fd, &read_fds))
	    {
		rc = recv(fd, tempbuf, 1, MSG_PEEK);
		if (rc == -1)
		{
		    /* -1 indicates ewouldblock (eagain) (check errno) */
		    p4_error("net_recv recv:  got -1", -1);
		}
		if (rc == 0)	/* if eof */
		{
		    /* eof; a process has closed its socket; may have died */
		    p4_error("net_recv recv:  EOF on socket", read_counter);
		}
		else
		    continue;
	    }
	    sleep(1);
	    if (eof_counter < 5)
		continue;
	    else
		p4_error("net_recv read:  probable EOF on socket fd", fd );
	}
#else
	{
	    /* Except on SYSV, n == 0 is EOF */
	    /* Note that this is an error even during rundown because sockets
	       should be closed with a "close socket" message first. */
	    p4_error("net_recv read:  probable EOF on socket", read_counter);
        }
#endif
	if (n < 0)
	{
	    /* EAGAIN is really POSIX, so we check for either EAGAIN 
	       or EWOULDBLOCK.  Note some systems set EAGAIN == EWOULDBLOCK
	     */
	    /* Solaris 2.5 occasionally sets n == -1 and errno == 0 (!!).
	       since n == -1 and errno == 0 is invalid (i.e., a bug in read),
	       it should be safe to treat it as EAGAIN and to try the
	       read once more (probably a race in the kernel)
	     */
	    if (errno == EAGAIN || errno == EWOULDBLOCK || errno == 0)
	    {
		COLLECT_STAT(n_recv_eagain++);
		block_counter++;
		/* Use a select here to wait for more data
		   to arrive.  This may give better performance, 
		   particularly when the system is actively involved in
		   trying to get the message to us
		 */
		if (p4_use_net_recv_w) {
		    fd_set         lread_fds;  /* l is for local */
		    struct timeval ltv;
		    int            ln1;
		    ltv.tv_sec = 5;     /* This is arbitrary */
		    ltv.tv_usec = 0;
		    FD_ZERO(&lread_fds);
		    FD_SET(fd, &lread_fds);
		    COLLECT_STAT(n_recv_select++);
		    SYSCALL_P4(ln1, select(fd+1, &lread_fds, 0, 0, &ltv));
		}
		else if (p4_use_readb && !set_fd_blocking) {
		    int flags;
		    set_fd_blocking = 1;
		    /* If we cached these flags in the p4 structure 
		       associated with the fd, we could avoid the F_GETFL */
		    flags = fcntl( fd, F_GETFL, 0 );
		    orig_flags = flags;
		    flags &= ~O_NDELAY;
		    fcntl( fd, F_SETFL, flags );
		}
		continue;
	    }
	    else {
		/* A closed socket can cause this to happen. */
		p4_dprintf( "net_recv failed for fd = %d\n", fd );
		p4_error("net_recv read, errno = ", errno);
	    }
	}
	recvd += n;
	COLLECT_STAT( if (n > n_recv_max) n_recv_max = n; );
	COLLECT_STAT( if (recvd < size) n_loop++);
    }
    p4_dprintfl( 99, 
		"Ending net_recv of %d on fd %d (eof_c = %d, block = %d)\n", 
		 size, fd, eof_counter, block_counter );
    COLLECT_STAT(if (n_loop > n_recv_maxloop) n_recv_maxloop = n_loop;);
    if (set_fd_blocking) fcntl( fd, F_SETFL, orig_flags );
    return (recvd);
}

/* flag --> fromid < toid; tie-breaker to avoid 2 procs rcving at same time */
/* typically set false for small internal messages, esp. when ids may not */
/*     yet be available */
/* set true for user msgs which may be quite large */
int net_send(int fd, P4VOID *in_buf, int size, int flag)
{
    struct p4_msg *dmsg;
    int n, sent = 0;
    int write_counter = 0;
    int block_counter = 0;
    char *buf = (char *)in_buf;

    /* net_send_w is a tuned version of net_send */
    if (p4_use_net_send_w) return net_send_w( fd, in_buf, size, flag );

    p4_dprintfl( 99, "Starting net_send of %d on fd %d\n", size, fd );
    while (sent < size)
    {
	write_counter++;		/* for debugging */
	SYSCALL_P4(n, write(fd, buf + sent, size - sent));
	if (n < 0)
	{
	    /* See net_read; these are often the same and EAGAIN is POSIX */
	    if (errno == EAGAIN || errno == EWOULDBLOCK)
	    {
/*	        
                p4_dprintfl( 90, "write net_send in EAGAIN with %d left", 
			     size ); 
*/
		block_counter++;
		if (flag)
		{
		    /* Someone may be writing to us ... */
		    if (socket_msgs_available())
		    {
			dmsg = socket_recv( P4_FALSE );
			/* close of a connection may return a null msg */
			if (dmsg) 
			    queue_p4_message(dmsg, 
					     p4_local->queued_messages);
		    }
		}
		continue;
	    }
	    else
	    {
		if (p4_local->in_wait_for_exit) {
		    /* Exit the while if we can't send a close message */
		    break;
		}
		p4_dprintf("net_send: could not write to fd=%d, errno = %d\n",
			   fd, errno);
		p4_error("net_send write", n);
	    }
	}
	sent += n;
    }
    p4_dprintfl( 99, "Ending net_send of %d on fd %d (blocked %d times)\n", 
		 size, fd, block_counter );
    return (sent);
}

/* net_send_w is a special version of net_send that uses select to wait on
 * *write* access to the socket as well as read access when a message cannot
 * be sent.  This keeps p4 from endless looping when it can't send. 
 */

int net_send_w(int fd, void *in_buf, int size, int flag)

/* flag --> fromid < toid; tie-breaker to avoid 2 procs rcving at same time */
/* typically set false for small internal messages, esp. when ids may not */
/*     yet be available */