p4_sock_util.c

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

/* set true for user msgs which may be quite large */
{
    struct p4_msg *dmsg;
    int n, sent = 0;
    int block_counter = 0;
    int size_left = size;
    char *buf = (char *)in_buf;
    COLLECT_STAT(int n_loop = 0);

    COLLECT_STAT(n_send_w_calls++);
    p4_dprintfl( 99, "Starting net_send_w of %d on fd %d\n", size, fd );
    while (size_left)
    {
	SYSCALL_P4(n, write(fd, buf + sent, size_left));
	if (n < 0)
	{
	    /* See net_read; these are often the same and EAGAIN is POSIX */
	    if (errno == EAGAIN || errno == EWOULDBLOCK)
	    {
		block_counter++;
		COLLECT_STAT(n_send_eagain++);
		/* Someone may be writing to us.  This waits until
		   either we can write or someone sends to use.
		   returns -1 if the write_fd is ready. */
		if (p4_sockets_ready( fd, 1 ) != -1)
		{
		    if (flag) {
			/* Only try to receive if the flag is set */
			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);
	    }
	}
	COLLECT_STAT(if (n >n_send_max) n_send_max = n);
	sent      += n;
	size_left -= n;
	COLLECT_STAT(if (size_left > 0) { n_send_looped++ ; n_loop++; });
    }
    p4_dprintfl( 99, "Ending net_send of %d on fd %d (blocked %d times)\n", 
		 size, fd, block_counter );
    COLLECT_STAT(if (n_loop > n_send_loopcnt) n_send_loopcnt = n_loop);
    return (sent);
}

/* Send the header and the message together if possible */
int net_send2( int fd, void *header, int header_len, 
	       void *data, int len, int flag )
{
    int n;

#if defined(HAVE_WRITEV) && defined(HAVE_SYS_UIO_H)
    if (p4_use_writev) {
	struct iovec vbuf[2];
	vbuf[0].iov_base = header;
	vbuf[0].iov_len  = header_len;
	vbuf[1].iov_base = data;
	vbuf[1].iov_len  = len;
	n = writev( fd, vbuf, 2 );
	if (n == -1) {
	    if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) {
		/* Just pretend nothing was written */
		n = 0;
	    }
	    else {
		/* Error on writing */
		/* We'll let the net_send code handle it */
		n = 0;
	    }
	}
	if (n < header_len + len) {
	    char *hptr = (char *)header;
	    if (n < header_len) {
		net_send( fd, hptr + n, header_len - n, flag );
		net_send( fd, data, len, flag );
	    }
	    else {
		char *dptr = (char *)data;
		int len_sent = n - header_len;
		net_send( fd, dptr + len_sent, len - len_sent, flag );
	    }
	}
	COLLECT_STAT(else {n_writev_first++;});
	/* Return only the length of the data sent */
	n = len;
    }
    else 
#endif
    {
	(void)net_send(fd, header, header_len, flag);
	p4_dprintfl(20, "sent hdr on fd %d via socket\n",fd);

	n = net_send(fd, data, len, flag);
    }
    return n;
}

void p4_socket_stat( FILE *fp )
{
    if (p4_output_socket_stat) {

	fprintf( fp, 
	     "send calls = %d eagain = %d maxbytes = %d loop %d maxloop %d\n",
	     n_send_w_calls, n_send_eagain, n_send_max, n_send_looped, 
	     n_send_loopcnt );
	fprintf( fp,
		 "send w writev %d\n", n_writev_first );
	fprintf( fp, 
	     "recv calls = %d eagain %d maxbytes %d select %d maxloop %d\n",
	     n_recv_calls, n_recv_eagain, n_recv_max, n_recv_select, 
	     n_recv_maxloop  );
	fflush( fp );
    }
}
/* This can FAIL if the host name is invalid.  For that reason, there is
   a timeout in the test, with a failure return if the entry cannot be found 

   Note also that the name returned may or may not be the canonnical, 
   "well known" name for the host, depending on the implementation of Unix.  
   This may not be the same as the input name, particularly 
   if the system has several networks.

   Finally, this can hang on systems that don't have a working name resolution
   service (this is not uncommon on LINUX clusters).  There is currently
   no fix for this (we need something like the timeout code in other 
   parts of the P4 implementation).

   We have added rudimentary timing to this routine to keep track of the
   amount of time that is spent in this routine.

   Another option, not implemented, is to maintain a local cache of
   names.  This would prevent us from making multiple queries about the same
   name.  However, since occurs most often when testing rather than using the
   p4 system, we have not implemented this idea.
 */
#include <sys/time.h>
static time_t time_in_gethostbyname = 0;
static int    n_gethostbyname = 0;
void p4_timein_hostbyname( int *t, int *count )
{
    *t     = (int)time_in_gethostbyname;
    *count = n_gethostbyname;
}
#ifndef TIMEOUT_VALUE 
#define TIMEOUT_VALUE 60
#endif
struct hostent *gethostbyname_p4(char *hostname)
{
    struct hostent *hp;
#ifdef SCYLD_BEOWULF
    struct sockaddr_in s_in;    
    long nodenum;
    int size;
    
    p4_dprintfl(10,"Beowulf: using beowulf version of gethostbyname_p4\n");
    
    nodenum=strtol(hostname,NULL,10);
    
    size=sizeof(struct sockaddr_in);
    bproc_nodeaddr(nodenum,(struct sockaddr *)&s_in,&size);
    
    hp=(struct hostent *)calloc(1,sizeof(struct hostent));
    hp->h_name=strdup(hostname);
    hp->h_aliases=NULL;
    hp->h_addrtype=AF_INET;
    hp->h_length=4;
    hp->h_addr_list=(char **)calloc(2,sizeof(char *));
    hp->h_addr_list[0]=calloc(1,4);
    memcpy(hp->h_addr_list[0],(char *)&(s_in.sin_addr.s_addr),4);
    hp->h_addr_list[1]=NULL;
#else
    int i = 100;
    time_t start_time, cur_time;

    start_time = time( (time_t) 0 );

    while ((hp = gethostbyname(hostname)) == NULL)
    {
	if (!--i)
	{
	    i = 100;
	    p4_dprintfl(10,"gethostbyname failed 100 times for host %s\n",
			hostname);
	    cur_time = time( (time_t) 0 );
	    if (cur_time - start_time > TIMEOUT_VALUE) {
		/* Dump out current procgroup */
		char msgbuf[512];
		if (p4_local && p4_local->procgroup) 
		    dump_procgroup(p4_local->procgroup,00);
		sprintf( msgbuf, 
"Could not gethostbyname for host %s; may be invalid name\n", hostname );
		p4_error(msgbuf, cur_time - start_time);
		return 0;
	    }
	}
    }
    time_in_gethostbyname += (time( (time_t) 0 ) - start_time );
    n_gethostbyname ++;
#endif /* SCYLD_BEOWULF */
    return(hp);
}

/* General replacement for gethostname for Solaris and Scyld */
int gethostname_p4(char *name,size_t len)
{
#   if defined(SUN_SOLARIS) || defined(MEIKO_CS2)
	return sysinfo(SI_HOSTNAME, name, len );
#   elif defined (SCYLD_BEOWULF)
	return -(snprintf(name, len , "%d", bproc_currnode()) == -1);
#   else
	return gethostname(name,len);
#   endif
}

P4VOID get_inet_addr(struct in_addr *addr)
{
    char hostname[100];
    struct hostent *hp;

    hostname[0] = '\0';
    get_qualified_hostname(hostname,sizeof(hostname));
    hp = gethostbyname_p4(hostname);
    bcopy(hp->h_addr, addr, hp->h_length);
}

P4VOID get_inet_addr_str( char *str )
{
    struct in_addr addr;

    get_inet_addr(&addr);
    strcpy(str, (char *) inet_ntoa(addr));
}

/* 
   This routine prints information on a socket, including many of the options
 */
void p4_print_sock_params( int skt )
{
#ifdef CAN_DO_SETSOCKOPT
    int rc, ival;
    p4_sockopt_len_t ivallen;
#ifdef SO_KEEPALIVE
    ivallen = sizeof(ival);
    rc = getsockopt( skt, SOL_SOCKET, SO_KEEPALIVE, (char *)&ival, &ivallen );
    if (!rc)
	printf( "Socket %d SO_KEEPALIVE = %d\n", skt, ival );
#endif
#ifdef SO_OOBINLINE
    ivallen = sizeof(ival);
    rc = getsockopt( skt, SOL_SOCKET, SO_OOBINLINE, (char *)&ival, &ivallen );
    if (!rc)
	printf( "Socket %d SO_OOBINLINE = %d\n", skt, ival );
#endif
#ifdef SO_SNDBUF
    ivallen = sizeof(ival);
    rc = getsockopt( skt, SOL_SOCKET, SO_SNDBUF, (char *)&ival, &ivallen );
    if (!rc)
	printf( "Socket %d SO_SNDBUF = %d\n", skt, ival );
#endif
#ifdef SO_RCVBUF
    ivallen = sizeof(ival);
    rc = getsockopt( skt, SOL_SOCKET, SO_RCVBUF, (char *)&ival, &ivallen );
    if (!rc)
	printf( "Socket %d SO_RCVBUF = %d\n", skt, ival );
#endif
#ifdef SO_SNDTIMEO
    ivallen = sizeof(ival);
    rc = getsockopt( skt, SOL_SOCKET, SO_SNDTIMEO, (char *)&ival, &ivallen );
    if (!rc)
	printf( "Socket %d SO_SNDTIMEO = %d\n", skt, ival );
#endif
#ifdef SO_RCVTIMEO
    ivallen = sizeof(ival);
    rc = getsockopt( skt, SOL_SOCKET, SO_RCVTIMEO, (char *)&ival, &ivallen );
    if (!rc)
	printf( "Socket %d SO_RCVTIMEO = %d\n", skt, ival );
#endif
#ifdef SO_SNDLOWAT
    ivallen = sizeof(ival);
    rc = getsockopt( skt, SOL_SOCKET, SO_SNDLOWAT, (char *)&ival, &ivallen );
    if (!rc)
	printf( "Socket %d SO_SNDLOWAT = %d\n", skt, ival );
#endif
#ifdef SO_RCVLOWAT
    ivallen = sizeof(ival);
    rc = getsockopt( skt, SOL_SOCKET, SO_RCVLOWAT, (char *)&ival, &ivallen );
    if (!rc)
	printf( "Socket %d SO_RCVLOWAT = %d\n", skt, ival );
#endif
#endif
}

#if !defined(CRAY)
/* cray complains about addr being addr of bit field */
/* can probably get around this problem if ever necessary */

P4VOID dump_sockaddr(char *who, struct sockaddr_in *sa)
{
    unsigned char *addr;

    addr = (unsigned char *) &(sa->sin_addr.s_addr);

    p4_dprintfl(90,"%s: family=%d port=%d addr=%d.%d.%d.%d\n",
		who,
                sa->sin_family,
                ntohs(sa->sin_port),
                addr[0], addr[1], addr[2], addr[3]);
}

P4VOID dump_sockinfo( char *msg, int fd)
{
    p4_sockopt_len_t nl;
    struct sockaddr_in peer, me;

    p4_dprintfl(00, "Dumping sockinfo for fd=%d: %s\n", fd, msg);

    nl = sizeof(me);
    getsockname(fd, (struct sockaddr *) &me, &nl);
    dump_sockaddr("Me", &me);
	   
    nl = sizeof(peer);
    getpeername(fd, (struct sockaddr *) &peer, &nl);
    dump_sockaddr("Peer",&peer);
}

#endif

/*
 * mpiexec_reopen_stdin is used by the OSC mpiexec interface.  This 
 * reinitializes stdin to a selected host and port.
 */

/*
 * Search the environment for variables which might say that mpiexec
 * requested stdin be grabbed from the spawning process.  Only happens
 * in the case of "-allstdin", i.e., where the user requested that the
 * same input be replicated into each process.
 */
void mpiexec_reopen_stdin(void)
{
    char *host = getenv("MPIEXEC_STDIN_HOST");
    char *sport = getenv("MPIEXEC_STDIN_PORT");
    struct sockaddr_in s_in;
    char *cq;
    int fd, port, tries;
    struct hostent *hp;

    if (!sport || !host)
	return;
    hp = gethostbyname_p4(host);
    if (!hp)
	p4_error("mpiexec_reopen_stdin: MPIEXEC_STDIN_HOST did not parse", 0);
    port = strtol(sport, &cq, 10);
    if (*cq)
	p4_error("mpiexec_reopen_stdin: MPIEXEC_STDIN_PORT did not parse", 0);
    fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0)
	p4_error("mpiexec_reopen_stdin: socket", fd);
    memset(&s_in, 0, sizeof(s_in));
    s_in.sin_family = AF_INET;
    s_in.sin_port = htons(port);
    memcpy(&s_in.sin_addr, hp->h_addr_list[0], hp->h_length);

    /*
     * Probably not necessary in the general case, but a swamped mpiexec
     * stdio process with a short listening backlog might require this.
     */
    tries = 0;
    for (;;) {
	int cc;

	cc = connect(fd, (struct sockaddr *)&s_in, sizeof(s_in));
	if (cc == 0)
	    break;
	if ((errno == ECONNREFUSED || errno == EINTR || errno == EAGAIN)
	  && tries < 5) {
	    ++tries;
	    sleep(1);
	    continue;
	}
	p4_error("mpiexec_reopen_stdin: connect", cc);
    }
    close(0);
    dup2(fd, 0);
    close(fd);
}

int p4_make_socket_nonblocking( int fd )
{
    /* Set the socket to be nonblocking.  */
    int rc, flags = fcntl( fd, F_GETFL, 0 );
    flags |= O_NONBLOCK;
    rc = fcntl( fd, F_SETFL, flags );
    return rc;
}