sock_post.i

mpi并行计算的c++代码 可用vc或gcc编译通过 可以用来搭建并行计算试验环境

/* -*- Mode: C; c-basic-offset:4 ; -*- */

/*
 *  (C) 2001 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */

#undef FUNCNAME 
#define FUNCNAME MPIDU_Sock_post_connect_ifaddr
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
/* 
 This routine connects to a particular address (in byte form; for ipv4, 
 the address is four bytes, typically the value of h_addr_list[0] in 
 struct hostent.  By avoiding a character name for an interface (we *never*
 connect to a host; we are *always* connecting to a particular interface 
 on a host), we avoid problems with DNS services, including lack of properly
 configured services and scalability problems.  As this routine uses 
 a four-byte field, it is currently restricted to ipv4.  This routine should
 evolve to support ipv4 and ipv6 addresses.

 This routine was constructed from MPIDU_Sock_post_connect by removing the 
 poorly placed use of gethostname within the middle of that routine and
 simply using the ifaddr field that is passed to this routine.  
 MPIDU_Sock_post_connect simply uses the hostname field to get the canonical
 IP address.  The original routine and its API was retained to allow backwards
 compatibility until it is determined that we can always use explicit addrs 
 needed in setting up the socket instead of character strings.
 */
int MPIDU_Sock_post_connect_ifaddr( struct MPIDU_Sock_set * sock_set, 
				    void * user_ptr, 
				    unsigned char ifaddr[], int port,
				    struct MPIDU_Sock ** sockp)
{
    struct MPIDU_Sock * sock = NULL;
    struct pollfd * pollfd;
    struct pollinfo * pollinfo;
    struct hostent * hostent;
    int fd = -1;
    struct sockaddr_in addr;
    long flags;
    int nodelay;
    int rc;
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPIDU_SOCK_POST_CONNECT);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDU_SOCK_POST_CONNECT);

    MPIDU_SOCKI_VERIFY_INIT(mpi_errno, fn_exit);

    /*
     * Create a non-blocking socket with Nagle's algorithm disabled
     */
    fd = socket(PF_INET, SOCK_STREAM, 0);
    if (fd == -1) {
	MPIU_ERR_SETANDJUMP2(mpi_errno,MPIDU_SOCK_ERR_FAIL,
			     "**sock|poll|socket", 
		    "**sock|poll|socket %d %s", errno, MPIU_Strerror(errno));
    }

    flags = fcntl(fd, F_GETFL, 0);
    if (flags == -1) {
	MPIU_ERR_SETANDJUMP2(mpi_errno,MPIDU_SOCK_ERR_FAIL,
			     "**sock|poll|nonblock", 
                    "**sock|poll|nonblock %d %s", errno, MPIU_Strerror(errno));
    }
    rc = fcntl(fd, F_SETFL, flags | O_NONBLOCK);
    if (rc == -1) {
	MPIU_ERR_SETANDJUMP2( mpi_errno, MPIDU_SOCK_ERR_FAIL,
			      "**sock|poll|nonblock", 
			      "**sock|poll|nonblock %d %s",
			      errno, MPIU_Strerror(errno));
    }

    nodelay = 1;
    rc = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &nodelay, sizeof(nodelay));
    if (rc != 0) {
	MPIU_ERR_SETANDJUMP2(mpi_errno,MPIDU_SOCK_ERR_FAIL,
			     "**sock|poll|nodelay", 
			     "**sock|poll|nodelay %d %s", 
			     errno, MPIU_Strerror(errno));
    }

    /*
     * Allocate and initialize sock and poll structures
     *
     * NOTE: pollfd->fd is initialized to -1.  It is only set to the true fd 
     * value when an operation is posted on the sock.  This
     * (hopefully) eliminates a little overhead in the OS and avoids 
     * repetitive POLLHUP events when the connection is closed by
     * the remote process.
     */
    mpi_errno = MPIDU_Socki_sock_alloc(sock_set, &sock);
    if (mpi_errno != MPI_SUCCESS) {
	MPIU_ERR_SETANDJUMP(mpi_errno,MPIDU_SOCK_ERR_NOMEM,
			    "**sock|sockalloc");
    }

    pollfd = MPIDU_Socki_sock_get_pollfd(sock);
    pollinfo = MPIDU_Socki_sock_get_pollinfo(sock);
    
    pollinfo->fd = fd;
    pollinfo->user_ptr = user_ptr;
    pollinfo->type = MPIDU_SOCKI_TYPE_COMMUNICATION;
    pollinfo->state = MPIDU_SOCKI_STATE_CONNECTED_RW;
    pollinfo->os_errno = 0;

    memset(&addr, 0, sizeof(addr));
    addr.sin_family = AF_INET;
    memcpy(&addr.sin_addr.s_addr, ifaddr, sizeof(addr.sin_addr.s_addr));
    addr.sin_port = htons(port);

    /*
     * Set and verify the socket buffer size
     */
    if (MPIDU_Socki_socket_bufsz > 0)
    {
	int bufsz;
	socklen_t bufsz_len;

	bufsz = MPIDU_Socki_socket_bufsz;
	bufsz_len = sizeof(bufsz);
	rc = setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &bufsz, bufsz_len);
	if (rc == -1) {
	    MPIU_ERR_SETANDJUMP3(mpi_errno,MPIDU_SOCK_ERR_FAIL, 
				 "**sock|poll|setsndbufsz",
				 "**sock|poll|setsndbufsz %d %d %s", 
				 bufsz, errno, MPIU_Strerror(errno));
	}
	bufsz = MPIDU_Socki_socket_bufsz;
	bufsz_len = sizeof(bufsz);
	rc = setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &bufsz, bufsz_len);
	if (rc == -1) {
	    MPIU_ERR_SETANDJUMP3(mpi_errno,MPIDU_SOCK_ERR_FAIL, 
				 "**sock|poll|setrcvbufsz",
				 "**sock|poll|setrcvbufsz %d %d %s", 
				 bufsz, errno, MPIU_Strerror(errno));
	}
	bufsz_len = sizeof(bufsz);
	/* FIXME: This should not be an error or even a warning if
	 we don't get the requested socket size */
	rc = getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &bufsz, &bufsz_len);
	/* --BEGIN ERROR HANDLING-- */
	if (rc == 0)
	{
	    if (bufsz < MPIDU_Socki_socket_bufsz * 0.9 || 
		bufsz < MPIDU_Socki_socket_bufsz * 1.0)
	    {
		MPIU_Msg_printf("WARNING: send socket buffer size differs from requested size (requested=%d, actual=%d)\n",
				MPIDU_Socki_socket_bufsz, bufsz);
	    }
	}
	/* --END ERROR HANDLING-- */

    	bufsz_len = sizeof(bufsz);
	rc = getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &bufsz, &bufsz_len);
	/* --BEGIN ERROR HANDLING-- */
	if (rc == 0)
	{
	    if (bufsz < MPIDU_Socki_socket_bufsz * 0.9 || 
		bufsz < MPIDU_Socki_socket_bufsz * 1.0)
	    {
		MPIU_Msg_printf("WARNING: receive socket buffer size differs from requested size (requested=%d, actual=%d)\n",
				MPIDU_Socki_socket_bufsz, bufsz);
	    }
	}
	/* --END ERROR HANDLING-- */
    }
    
    /*
     * Attempt to establish the connection
     */
    do
    {
        rc = connect(fd, (struct sockaddr *) &addr, sizeof(addr));
    }
    while (rc == -1 && errno == EINTR);
    
    if (rc == 0)
    {
	/* connection succeeded */
	pollinfo->state = MPIDU_SOCKI_STATE_CONNECTED_RW;
	MPIDU_SOCKI_EVENT_ENQUEUE(pollinfo, MPIDU_SOCK_OP_CONNECT, 0, user_ptr, MPI_SUCCESS, mpi_errno, fn_fail);
    }
    /* --BEGIN ERROR HANDLING-- */
    else if (errno == EINPROGRESS)
    {
	/* connection pending */
	pollinfo->state = MPIDU_SOCKI_STATE_CONNECTING;
	MPIDU_SOCKI_POLLFD_OP_SET(pollfd, pollinfo, POLLOUT);
    }
    else
    {
	pollinfo->os_errno = errno;
	pollinfo->state = MPIDU_SOCKI_STATE_DISCONNECTED;

	if (errno == ECONNREFUSED)
	{
	    MPIDU_SOCKI_EVENT_ENQUEUE(pollinfo, MPIDU_SOCK_OP_CONNECT, 0, user_ptr, MPIR_Err_create_code(
		MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_CONN_FAILED,
		"**sock|connrefused", "**sock|poll|connrefused %d %d %s",
		pollinfo->sock_set->id, pollinfo->sock_id, ""), mpi_errno, fn_fail);
	}
	else
	{
	    MPIDU_SOCKI_EVENT_ENQUEUE(pollinfo, MPIDU_SOCK_OP_CONNECT, 0, user_ptr, MPIR_Err_create_code(
		MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_CONN_FAILED,
		"**sock|oserror", "**sock|poll|oserror %d %d %d %s", pollinfo->sock_set->id, pollinfo->sock_id, errno,
		MPIU_Strerror(errno)), mpi_errno, fn_fail);
	}
    }
    /* --END ERROR HANDLING-- */

    *sockp = sock;

  fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPIDU_SOCK_POST_CONNECT);
    return mpi_errno;

    /* --BEGIN ERROR HANDLING-- */
  fn_fail:
    if (fd != -1)
    { 
	close(fd);
    }

    if (sock != NULL)
    {
	MPIDU_Socki_sock_free(sock);
    }

    goto fn_exit;
    /* --END ERROR HANDLING-- */
}

/* FIXME: What does this routine do?  Why does it take a host description
   instead of an interface name or address? */
#undef FUNCNAME
#define FUNCNAME MPIDU_Sock_post_connect
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
int MPIDU_Sock_post_connect(struct MPIDU_Sock_set * sock_set, void * user_ptr, 
			    char * host_description, int port,
			    struct MPIDU_Sock ** sockp)
{
    int mpi_errno = MPI_SUCCESS;
    struct hostent * hostent;

    /*
     * Convert hostname to IP address
     *
     * FIXME: this should handle failures caused by a backed up listener queue
     * at the remote process.  It should also use a
     * specific interface if one is specified by the user.
     */
    /* FIXME: strtok may change the contents of host_description.  Shouldn't
       the host description be a const char [] and not modified by this 
       routine? */
    strtok(host_description, " ");
    /* FIXME: For ipv6, we should use getaddrinfo */
    hostent = gethostbyname(host_description);
    /* --BEGIN ERROR HANDLING-- */
    if (hostent == NULL || hostent->h_addrtype != AF_INET) {
	/* FIXME: Set error */
	goto fn_exit;
    }
    /* --END ERROR HANDLING-- */
    mpi_errno = MPIDU_Sock_post_connect_ifaddr( sock_set, user_ptr, 
			(unsigned char *)hostent->h_addr_list[0], port, 
			sockp );
 fn_exit:
    return mpi_errno;
}
/* end MPIDU_Sock_post_connect() */


#undef FUNCNAME
#define FUNCNAME MPIDU_Sock_listen
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
#ifndef USHRT_MAX
#define USHRT_MAX 65535   /* 2^16-1 */
#endif
int MPIDU_Sock_listen(struct MPIDU_Sock_set * sock_set, void * user_ptr, 
		      int * port, struct MPIDU_Sock ** sockp)
{
    struct MPIDU_Sock * sock;
    struct pollfd * pollfd;
    struct pollinfo * pollinfo;
    int fd = -1;
    long flags;
    int optval;
    struct sockaddr_in addr;
    socklen_t addr_len;
    int rc;
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_MPIDU_SOCK_LISTEN);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDU_SOCK_LISTEN);

    MPIDU_SOCKI_VERIFY_INIT(mpi_errno, fn_exit);
    /* --BEGIN ERROR HANDLING-- */
    if (*port < 0 || *port > USHRT_MAX)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_BAD_PORT,
					 "**sock|badport", "**sock|badport %d", *port);
	goto fn_exit;
    }
    /* --END ERROR HANDLING-- */

    /*
     * Create a non-blocking socket for the listener
     */
    fd = socket(PF_INET, SOCK_STREAM, 0);
    /* --BEGIN ERROR HANDLING-- */
    if (fd == -1)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_FAIL,
					 "**sock|poll|socket", "**sock|poll|socket %d %s", errno, MPIU_Strerror(errno));
	goto fn_fail;
    }
    /* --END ERROR HANDLING-- */

    /* set SO_REUSEADDR to a prevent a fixed service port from being bound to during subsequent invocations */
    if (*port != 0)
    {
	optval = 1;
	rc = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(int));
	/* --BEGIN ERROR HANDLING-- */
	if (rc == -1)
	{
	    mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_FAIL,
					     "**sock|poll|reuseaddr", "**sock|poll|reuseaddr %d %s", errno, MPIU_Strerror(errno));
	    goto fn_fail;
	}
	/* --END ERROR HANDLING-- */
    }

    /* make the socket non-blocking so that accept() will return immediately if no new connection is available */
    flags = fcntl(fd, F_GETFL, 0);
    /* --BEGIN ERROR HANDLING-- */
    if (flags == -1)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_FAIL,
					 "**sock|poll|nonblock", "**sock|poll|nonblock %d %s", errno, MPIU_Strerror(errno));
	goto fn_fail;
    }
    /* --END ERROR HANDLING-- */
    rc = fcntl(fd, F_SETFL, flags | O_NONBLOCK);
    /* --BEGIN ERROR HANDLING-- */
    if (rc == -1)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_FAIL,
					 "**sock|poll|nonblock", "**sock|poll|nonblock %d %s", errno, MPIU_Strerror(errno));
	goto fn_fail;
    }
    /* --END ERROR HANDLING-- */

    /*
     * Bind the socket to all interfaces and the specified port.  The port specified by the calling routine may be 0, indicating
     * that the operating system can select an available port in the ephemeral port range.
     */
    if (*port == 0) {
	int portnum, low_port, high_port;
	/* see if we actually want to find values within a range */
	
	low_port = high_port = 0;
	/* Get range here.  These leave low_port, high_port unchanged
	   if the env variable is not set */
	MPIU_GetEnvRange( "MPICH_PORT_RANGE", &low_port, &high_port );

	for (portnum=low_port; portnum<=high_port; portnum++) {
	    memset( (void *)&addr, 0, sizeof(addr) );
	    addr.sin_family = AF_INET;
	    addr.sin_addr.s_addr = htonl(INADDR_ANY);
	    addr.sin_port	   = htons( portnum );
	    
	    rc = bind(fd, (struct sockaddr *) &addr, sizeof(addr));
	    if (rc < 0) {
		if (errno != EADDRINUSE && errno != EADDRNOTAVAIL) {
		    close(fd);
		    break;
		}
	    }
	    else 
		break;
	}
    }
    else {
	memset(&addr, 0, sizeof(addr));
	addr.sin_family = AF_INET;
	addr.sin_addr.s_addr = htonl(INADDR_ANY);
	addr.sin_port = htons((unsigned short) *port);
	rc = bind(fd, (struct sockaddr *) &addr, sizeof(addr));
    }
    /* --BEGIN ERROR HANDLING-- */
    if (rc == -1)
    {
	mpi_errno = MPIR_Err_create_code(MPI_SUCCESS, MPIR_ERR_RECOVERABLE, FCNAME, __LINE__, MPIDU_SOCK_ERR_FAIL,
					 "**sock|poll|bind", "**sock|poll|bind %d %d %s", port, errno, MPIU_Strerror(errno));
	goto fn_fail;
    }
    /* --END ERROR HANDLING-- */

    /*
     * Set and verify the socket buffer size
     */
    if (MPIDU_Socki_socket_bufsz > 0)