ch3u_connect_sock.c

fortran并行计算包

	    }
	}
    }

 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_CH3_SOCKCONN_HANDLE_CONNWRITE);
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}

/* ----------------------------------------------------------------------- */
/* FIXME: What does this do? */
#undef FUNCNAME
#define FUNCNAME MPIDI_CH3I_VC_post_sockconnect
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_CH3I_VC_post_sockconnect(MPIDI_VC_t * vc)
{
    int mpi_errno = MPI_SUCCESS;
    char val[MPIDI_MAX_KVS_VALUE_LEN];
    MPIDI_CH3I_VC *vcch = (MPIDI_CH3I_VC *)vc->channel_private;
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_CH3I_VC_POST_SOCKCONNECT);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_CH3I_VC_POST_SOCKCONNECT);

    /* MPIDI_PG_GetConnString() can block & release the lock for 
     * the current thread. Prevent other threads from trying to
     * obtain the ConnString by setting the VC to *CONNECTING.
     */
    if(vcch->state == MPIDI_CH3I_VC_STATE_UNCONNECTED){ 
    	MPIU_DBG_VCCHSTATECHANGE(vc,VC_STATE_CONNECTING);
    	vcch->state = MPIDI_CH3I_VC_STATE_CONNECTING;
	MPIU_DBG_MSG_P(CH3_CONNECT,TYPICAL,"vc=(%p) Going ahead to obtain connstring", vc);
    }else{
	MPIU_DBG_MSG_P(CH3_CONNECT,TYPICAL,"MT: vc=(%p) is already connecting/ed", vc);
	MPIU_DBG_MSG(CH3_CONNECT,TYPICAL,"Aborting posting a connect");
	/*************** MT *****************/
	/* There are 3 cases here,
         * 1) Another thread posted a connect while the current thread
         *    was blocked in MPIDI_PG_GetConnString()
         *    VC state = MPIDI_CH3I_VC_STATE_CONNECTING
         * 2) Another thread posted a connect and completed the 
         *    connection while the current thread was blocked in 
         *    MPIDI_PG_GetConnString()
         *    VC state = MPIDI_CH3I_VC_STATE_CONNECTED
         * 3) Another thread received a connect from the same proc we
         *    are connecting to and opened a connection while the 
         *    current thread was blocked in MPIDI_PG_GetConnString()
         *    VC state = MPIDI_CH3I_VC_STATE_CONNECTING or
         *    VC state = MPIDI_CH3I_VC_STATE_CONNECTED
         * If we bail out here, in all the cases above the other thread
         * will handle the connection. In particular in the 3rd case
         * if we proceed to post a connect before the VC state is set
         * by the thread processing the remote connect,
         * the code for head-to-head conn resolution will take care of
         * discarding one of the connections
         */
	 mpi_errno = MPI_SUCCESS;
         goto fn_exit;
    }
    mpi_errno = MPIDI_PG_GetConnString( vc->pg, vc->pg_rank, val, sizeof(val));
    if (mpi_errno != MPI_SUCCESS) {
	MPIU_ERR_POP(mpi_errno);
    }

    mpi_errno = MPIDI_CH3I_Sock_connect( vc, val, sizeof(val) );

  fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_CH3I_VC_POST_SOCKCONNECT);
    return mpi_errno;
 fn_fail:
    goto fn_exit;
    /* --END ERROR HANDLING-- */
}
/* end MPIDI_CH3I_VC_post_sockconnect() */

/* Given a connection string, start the process of creating a socket 
   connection to that designated interface (on a node).  This routine 
   is used both in MPIDI_CH3I_VC_post_sockconnect and in 
   MPIDI_CH3I_VC_post_connect in the ch3:ssm channel. 

   vallen = sizeof(val)
*/
#undef FUNCNAME
#define FUNCNAME MPIDI_CH3I_Sock_connect
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_CH3I_Sock_connect( MPIDI_VC_t *vc, const char val[], int vallen )
{
    char host_description[MAX_HOST_DESCRIPTION_LEN];
    MPIDU_Sock_ifaddr_t ifaddr;
    int hasIfaddr = 0, port;
    MPIDI_CH3I_Connection_t * conn = 0;
    int mpi_errno = MPI_SUCCESS;
    MPIDI_CH3I_VC *vcch = (MPIDI_CH3I_VC *)vc->channel_private;
    MPIDI_STATE_DECL(MPID_STATE_MPIDI_CH3I_SOCK_CONNECT);

    MPIDI_FUNC_ENTER(MPID_STATE_MPIDI_CH3I_SOCK_CONNECT);
    
    if(vcch->state == MPIDI_CH3I_VC_STATE_CONNECTING){ 
	MPIU_DBG_MSG_P(CH3_CONNECT,TYPICAL,"Posting a connect for vc=(%p)", vc);
    }else{
	MPIU_DBG_MSG_P(CH3_CONNECT,TYPICAL,"MT: vc=(%p) is already connected", vc);
	MPIU_DBG_MSG(CH3_CONNECT,TYPICAL,"Aborting posting a connect");
	/*************** MT *****************/
        /* 1) Another thread received a connect from the same proc
         *    the current thread is connecting to and opened a 
	 *    connection while the current thread was blocked in 
	 *    MPIDI_PG_GetConnString()
         *    VC state = MPIDI_CH3I_VC_STATE_CONNECTED
         * If we bail out here, the other thread will handle the connection. 
         * if we proceed to post a connect before the VC state is set
         * by the thread processing the remote connect,
         * the code for head-to-head conn resolution will take care of
         * discarding one of the connections
         */
	 mpi_errno = MPI_SUCCESS;
         goto fn_exit;
    }

    mpi_errno = MPIDU_Sock_get_conninfo_from_bc( val, host_description,
						 sizeof(host_description),
						 &port, &ifaddr, &hasIfaddr );
    if (mpi_errno) {
	MPIU_ERR_POP(mpi_errno);
    }

    mpi_errno = MPIDI_CH3I_Connection_alloc(&conn);
    if (mpi_errno == MPI_SUCCESS)
    {
	/* FIXME: This is a hack to allow Windows to continue to use
	   the host description string instead of the interface address
	   bytes when posting a socket connection.  This should be fixed 
	   by changing the Sock_post_connect to only accept interface
	   address.  See also channels/ssm/ch3_progress_connect.c */
#ifndef HAVE_WINDOWS_H
	if (hasIfaddr) {
	    mpi_errno = MPIDU_Sock_post_connect_ifaddr(MPIDI_CH3I_sock_set, 
						       conn, &ifaddr, port, 
						       &conn->sock);
	}
	else 
#endif
	{
	    mpi_errno = MPIDU_Sock_post_connect(MPIDI_CH3I_sock_set, conn, 
						host_description, port, 
						&conn->sock);
	}
	if (mpi_errno == MPI_SUCCESS)
	{
	    MPIU_DBG_CONNSTATECHANGE(vc,conn,CONN_STATE_CONNECTING);
	    vcch->sock = conn->sock;
	    vcch->conn = conn;
	    conn->vc = vc;
	    conn->state = CONN_STATE_CONNECTING;
	    conn->send_active = NULL;
	    conn->recv_active = NULL;
	}
	/* --BEGIN ERROR HANDLING-- */
	else
	{
	    MPIU_DBG_VCCHSTATECHANGE(vc,VC_STATE_FAILED);
	    vcch->state = MPIDI_CH3I_VC_STATE_FAILED;
	    mpi_errno = MPIR_Err_create_code(mpi_errno, MPIR_ERR_FATAL, FCNAME, __LINE__, MPI_ERR_OTHER, "**ch3|sock|postconnect",
		"**ch3|sock|postconnect %d %d %s", MPIR_Process.comm_world->rank, vc->pg_rank, val);
	    goto fn_fail;
	}
	/* --END ERROR HANDLING-- */
    }
    else {
	MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**ch3|sock|connalloc");
    }

 fn_exit:
    MPIDI_FUNC_EXIT(MPID_STATE_MPIDI_CH3I_SOCK_CONNECT);
    return mpi_errno;
 fn_fail:
    /* --BEGIN ERROR HANDLING-- */
    if (conn) {
	connection_destroy(conn);
    }
    goto fn_exit;
    /* --END ERROR HANDLING-- */
}


/* FIXME: What does this do? */
/* Guess: Setup a wait-to-read on the socket that was set after the accept 
   was handled */
/* Wrong guess.  */
#undef FUNCNAME
#define FUNCNAME connection_post_recv_pkt
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int connection_post_recv_pkt(MPIDI_CH3I_Connection_t * conn)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_CONNECTION_POST_RECV_PKT);

    MPIDI_FUNC_ENTER(MPID_STATE_CONNECTION_POST_RECV_PKT);

    mpi_errno = MPIDU_Sock_post_read(conn->sock, &conn->pkt, sizeof(conn->pkt),
				     sizeof(conn->pkt), NULL);
    if (mpi_errno != MPI_SUCCESS) {
	MPIU_ERR_POP(mpi_errno);
    }

 fn_fail:    
    MPIDI_FUNC_EXIT(MPID_STATE_CONNECTION_POST_RECV_PKT);
    return mpi_errno;
}


#undef FUNCNAME
#define FUNCNAME connection_post_send_pkt
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int connection_post_send_pkt(MPIDI_CH3I_Connection_t * conn)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_STATE_DECL(MPID_STATE_CONNECTION_POST_SEND_PKT);

    MPIDI_FUNC_ENTER(MPID_STATE_CONNECTION_POST_SEND_PKT);
 
    MPIU_DBG_PKT(conn,&conn->pkt,"connect");
    mpi_errno = MPIDU_Sock_post_write(conn->sock, &conn->pkt, sizeof(conn->pkt),
				      sizeof(conn->pkt), NULL);
    if (mpi_errno != MPI_SUCCESS) {
	MPIU_ERR_POP(mpi_errno);
    }
    
 fn_fail:
    MPIDI_FUNC_EXIT(MPID_STATE_CONNECTION_POST_SEND_PKT);
    return mpi_errno;
}

#undef FUNCNAME
#define FUNCNAME connection_post_send_pkt_and_pgid
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int connection_post_send_pkt_and_pgid(MPIDI_CH3I_Connection_t * conn)
{
    int mpi_errno;
    MPIDI_STATE_DECL(MPID_STATE_CONNECTION_POST_SEND_PKT_AND_PGID);

    MPIDI_FUNC_ENTER(MPID_STATE_CONNECTION_POST_SEND_PKT_AND_PGID);
    
    conn->iov[0].MPID_IOV_BUF = (MPID_IOV_BUF_CAST) &conn->pkt;
    conn->iov[0].MPID_IOV_LEN = (int) sizeof(conn->pkt);

    conn->iov[1].MPID_IOV_BUF = (MPID_IOV_BUF_CAST) MPIDI_Process.my_pg->id;
    conn->iov[1].MPID_IOV_LEN = (int) strlen(MPIDI_Process.my_pg->id) + 1;

    MPIU_DBG_PKT(conn,&conn->pkt,"connect-pgid");
    mpi_errno = MPIDU_Sock_post_writev(conn->sock, conn->iov, 2, NULL);
    if (mpi_errno != MPI_SUCCESS) {
	MPIU_ERR_POP(mpi_errno);
    }
    
 fn_fail:
    MPIDI_FUNC_EXIT(MPID_STATE_CONNECTION_POST_SEND_PKT_AND_PGID);
    return mpi_errno;
}

/* FIXME: This function also used in channels/sock/src/ch3_progress.c */
#undef FUNCNAME
#define FUNCNAME connection_post_sendq_req
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static int connection_post_sendq_req(MPIDI_CH3I_Connection_t * conn)
{
    int mpi_errno = MPI_SUCCESS;
    MPIDI_CH3I_VC *vcch = (MPIDI_CH3I_VC *)conn->vc->channel_private;
    MPIDI_STATE_DECL(MPID_STATE_CONNECTION_POST_SENDQ_REQ);

    MPIDI_FUNC_ENTER(MPID_STATE_CONNECTION_POST_SENDQ_REQ);

    /* post send of next request on the send queue */
    conn->send_active = MPIDI_CH3I_SendQ_head(vcch); /* MT */
    if (conn->send_active != NULL)
    {
	MPIU_DBG_MSG_P(CH3_CONNECT,TYPICAL,"conn=%p: Posting message from connection send queue", conn );
	mpi_errno = MPIDU_Sock_post_writev(conn->sock, 
					   conn->send_active->dev.iov, 
					   conn->send_active->dev.iov_count, 
					   NULL);
	if (mpi_errno != MPI_SUCCESS) {
	    MPIU_ERR_POP(mpi_errno);
	}
    }
    
 fn_fail:
    MPIDI_FUNC_EXIT(MPID_STATE_CONNECTION_POST_SENDQ_REQ);
    return mpi_errno;
}


/* This routine frees all of the memory associated with a connection.
   It is named destroy instead of free because routines with name "free" 
   should have MPI semantics - free means to 
   decrement reference count and free if reference count is zero */
#undef FUNCNAME
#define FUNCNAME connection_destroy
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
static void connection_destroy(MPIDI_CH3I_Connection_t * conn)
{
    MPIDI_STATE_DECL(MPID_STATE_CONNECTION_DESTROY);

    MPIDI_FUNC_ENTER(MPID_STATE_CONNECTION_DESTROY);

    MPIU_Free(conn->pg_id);
    MPIU_Free(conn);
    
    MPIDI_FUNC_EXIT(MPID_STATE_CONNECTION_DESTROY);
}


#ifdef USE_DBG_LOGGING
const char * MPIDI_CH3_VC_SockGetStateString( struct MPIDI_VC *vc )
{
    const char *name = "unknown";
    static char asdigits[20];
    MPIDI_CH3I_VC *vcch = (MPIDI_CH3I_VC *)vc->channel_private;
    int    state = vcch->state;
    
    switch (state) {
    case MPIDI_CH3I_VC_STATE_UNCONNECTED: name = "CH3I_VC_STATE_UNCONNECTED"; break;
    case MPIDI_CH3I_VC_STATE_CONNECTING:  name = "CH3I_VC_STATE_CONNECTING"; break;
    case MPIDI_CH3I_VC_STATE_CONNECTED:   name = "CH3I_VC_STATE_CONNECTED"; break;
    case MPIDI_CH3I_VC_STATE_FAILED:      name = "CH3I_VC_STATE_FAILED"; break;
    default:
	MPIU_Snprintf( asdigits, sizeof(asdigits), "%d", state );
	asdigits[20-1] = 0;
	name = (const char *)asdigits;
    }
    return name;
}
#endif