sctp_util.c

fortran并行计算包

    if(iovp-> MPID_IOV_LEN == 0)
      continue;

    read_size = MPIR_MIN(bytes_read, iovp-> MPID_IOV_LEN);

    MEMCPY(iovp-> MPID_IOV_BUF, from, read_size);

    from += read_size;
    bytes_read -= read_size;
  }

  return rc;
}

#undef FUNCNAME
#define FUNCNAME Req_Stream_from_pkt_and_req
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
int Req_Stream_from_pkt_and_req(MPIDI_CH3_Pkt_t * pkt, MPID_Request * sreq)
{
    int stream;
    
    switch(pkt->type)
    {
        /* account for each MPIDI_CH3_Pkt_type */
    case MPIDI_CH3_PKT_EAGERSHORT_SEND:

        /* these types are internally identical */
        case MPIDI_CH3_PKT_EAGER_SEND :
        case MPIDI_CH3_PKT_EAGER_SYNC_SEND :
        case MPIDI_CH3_PKT_READY_SEND :
        case MPIDI_CH3_PKT_CANCEL_SEND_REQ:
        case MPIDI_CH3_PKT_RNDV_REQ_TO_SEND :
        {
	  stream = Req_Stream_from_match(pkt->eager_send.match);
        }
        break;
        
        case MPIDI_CH3_PKT_RNDV_SEND :
        { 
	  stream = Req_Stream_from_match(sreq->dev.match);
        }
        break;
        
        default :
        case MPIDI_CH3_PKT_CLOSE :
        case MPIDI_CH3_PKT_RNDV_CLR_TO_SEND :    /*  CTS has unset values here */
                                                 /*  FIXME : need to be smart about assigning the CTS stream
                                                  *    so it doesn't overlap an already active stream
                                                  */
        {
	  stream = MPICH_SCTP_CTL_STREAM;
        }
        break;
    }
    MPIU_DBG_MSG_D(CH3_CONNECT,VERBOSE,"stream %d returned",stream );

    return stream;        
}


#undef FUNCNAME
#define FUNCNAME MPIDU_Sctp_post_writev
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
inline int MPIDU_Sctp_post_writev(MPIDI_VC_t* vc, MPID_Request* sreq, int offset,
				  MPIDU_Sock_progress_update_func_t fn, int stream_no)
{
  MPIU_Assert(offset >= 0);

  int mpi_errno = MPI_SUCCESS;
  MPID_IOV* iov = sreq->dev.iov + offset;
  int iov_n = sreq->dev.iov_count - offset;
  
  Global_SendQ_enqueue(vc, sreq, stream_no);
  
  SCTP_IOV* iov_ptr = &(vc->ch.posted_iov[stream_no]);
  
  POST_IOV(iov_ptr) = iov;
  POST_IOV_CNT(iov_ptr) = iov_n;
  POST_IOV_OFFSET(iov_ptr) = 0;
  POST_IOV_FLAG(iov_ptr) = TRUE;
  
  /* POST_UPDATE_FN(iov_ptr) = fn; */
  
 fn_exit:
  return mpi_errno;
}


#undef FUNCNAME
#define FUNCNAME MPIDU_Sctp_post_write
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
inline int MPIDU_Sctp_post_write(MPIDI_VC_t* vc, MPID_Request* sreq, MPIU_Size_t minlen, 
			  MPIU_Size_t maxlen, MPIDU_Sock_progress_update_func_t fn, int stream_no) {
  
  int mpi_errno = MPI_SUCCESS;
  void* buf = sreq->dev.iov[0].MPID_IOV_BUF;
  
  Global_SendQ_enqueue(vc, sreq, stream_no);

  SCTP_IOV* iov_ptr = &(vc->ch.posted_iov[stream_no]);

  POST_IOV_FLAG(iov_ptr) = FALSE;
  POST_BUF(iov_ptr) = buf;
  POST_BUF_MIN(iov_ptr) = minlen;
  POST_BUF_MAX(iov_ptr) = maxlen;
  
  /* POST_UPDATE_FN(iov_ptr) = fn; */

 fn_exit:
  return mpi_errno;
}


/* BUFFER Management routines */

static BufferNode_t* BufferList = NULL;

#undef FUNCNAME
#define FUNCNAME BufferList_init
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
inline void BufferList_init(BufferNode_t* node) {
  BufferList = node;
  buf_init(BufferList);
}

#undef FUNCNAME
#define FUNCNAME buf_init
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
inline int buf_init(BufferNode_t* node) {
  node-> free_space = READ_AMOUNT;
  node-> buf_ptr = node-> buffer;

  node-> next = NULL;
  node-> dynamic = FALSE;

  return MPI_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME buf_clean
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
inline int buf_clean(BufferNode_t* node) {
  return MPI_SUCCESS;
}

#undef FUNCNAME
#define FUNCNAME request_buffer
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
inline char* request_buffer(int size, BufferNode_t** bf_node) {
  BufferNode_t* node = BufferList;
  if(node-> free_space >= size) {
    *bf_node = node;
    return node-> buf_ptr;
  } else { 
    *bf_node = NULL;
    return NULL;
  }
}

#undef FUNCNAME
#define FUNCNAME update_size
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
inline void update_size(BufferNode_t* node, int size) {
  node-> free_space -= size;
  node-> buf_ptr += size;
}

/* END Buffer routines */


/* got rid of sock util so need sctp versions of these definitions */


#define MPIDI_CH3I_HOST_DESCRIPTION_KEY  "description"
#define MPIDI_CH3I_PORT_KEY              "port"
#define MPIDI_CH3I_IFNAME_KEY            "ifname"


#undef FUNCNAME
#define FUNCNAME MPIDU_Sctp_get_host_description
#undef FCNAME
#define FCNAME MPIU_QUOTE(FUNCNAME)
static int MPIDU_Sctp_get_host_description(char * host_description, int len)
{
    char * env_hostname;
    int rc;
    int mpi_errno = MPI_SUCCESS;
   
    /* --BEGIN ERROR HANDLING-- */
    if (len < 0)
    {
      mpi_errno = -1;
      goto fn_exit;
    }
    /* --END ERROR HANDLING-- */

    /* FIXME: Is this documented?  How does it work if the process manager
       cannot give each process a different value for an environment
       name?  What if a different interface is needed? */
    /* Use hostname supplied in environment variable, if it exists */
    env_hostname = getenv("MPICH_INTERFACE_HOSTNAME");
#if 0
    if (!env_hostname) {
	/* FIXME: Try to get the environment variable that uses the rank 
	   in comm world, i.e., MPICH_INTERFACE_HOSTNAME_R_%d.  For 
	   this, we'll need to know the rank for this process. */
    }
#endif
    if (env_hostname != NULL)
    {
	rc = MPIU_Strncpy(host_description, env_hostname, len);
	/* --BEGIN ERROR HANDLING-- */
	if (rc != 0)
	{
	  mpi_errno = -1;
	}
	/* --END ERROR HANDLING-- */
    }
    else {
	rc = gethostname(host_description, len);
	/* --BEGIN ERROR HANDLING-- */
	if (rc == -1)
	{
	  mpi_errno = -1;
	}
	/* --END ERROR HANDLING-- */
    }

  fn_exit:
    return mpi_errno;
}



#undef FUNCNAME
#define FUNCNAME MPIDI_CH3U_Get_business_card_sctp
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDI_CH3U_Get_business_card_sctp(char **bc_val_p, int *val_max_sz_p)
{
    int mpi_errno = MPI_SUCCESS;
    int port;
    char host_description[MAX_HOST_DESCRIPTION_LEN];
    
    mpi_errno = MPIDU_Sctp_get_host_description(host_description, MAX_HOST_DESCRIPTION_LEN);
    if (mpi_errno != MPI_SUCCESS) {
	MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**init_description");
    }

    port = MPIDI_CH3I_listener_port;
    mpi_errno = MPIU_Str_add_int_arg(bc_val_p, val_max_sz_p, MPIDI_CH3I_PORT_KEY, port);
    /* --BEGIN ERROR HANDLING-- */
    if (mpi_errno != MPIU_STR_SUCCESS)
    {
	if (mpi_errno == MPIU_STR_NOMEM) {
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**buscard_len");
	}
	else {
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**buscard");
	}
    }
    /* --END ERROR HANDLING-- */
    
    mpi_errno = MPIU_Str_add_string_arg(bc_val_p, val_max_sz_p, MPIDI_CH3I_HOST_DESCRIPTION_KEY, host_description);
    /* --BEGIN ERROR HANDLING-- */
    if (mpi_errno != MPIU_STR_SUCCESS)
    {
	if (mpi_errno == MPIU_STR_NOMEM) {
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**buscard_len");
	}
	else {
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**buscard");
	}
	return mpi_errno;
    }
    /* --END ERROR HANDLING-- */

    /* Look up the interface address cooresponding to this host description */
    /* FIXME: We should start switching to getaddrinfo instead of 
       gethostbyname */
    /* FIXME: We don't make use of the ifname in Windows in order to 
       provide backward compatibility with the (undocumented) host
       description string used by the socket connection routine 
       MPIDU_Sock_post_connect.  We need to change to an interface-address
       (already resolved) based description for better scalability and
       to eliminate reliance on fragile DNS services. Note that this is
       also more scalable, since the DNS server may serialize address 
       requests.  On most systems, asking for the host info of yourself
       is resolved locally (i.e., perfectly parallel).
    */
#ifndef HAVE_WINDOWS_H
    {
	struct hostent *info;
	char ifname[256];
	unsigned char *p;
	info = gethostbyname( host_description );
	if (info && info->h_addr_list) {
	    p = (unsigned char *)(info->h_addr_list[0]);
	    MPIU_Snprintf( ifname, sizeof(ifname), "%u.%u.%u.%u", 
			   p[0], p[1], p[2], p[3] );
	    MPIU_DBG_MSG_S(CH3_CONNECT,VERBOSE,"ifname = %s",ifname );
	    mpi_errno = MPIU_Str_add_string_arg( bc_val_p, 
						 val_max_sz_p, 
						 MPIDI_CH3I_IFNAME_KEY,
						 ifname );
	    if (mpi_errno != MPIU_STR_SUCCESS) {
		if (mpi_errno == MPIU_STR_NOMEM) {
		    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**buscard_len");
		}
		else {
		    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**buscard");
		}
	    }
	}
    }
#endif
 fn_exit:
    
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}


#undef FUNCNAME
#define FUNCNAME MPIDU_Sctp_get_conninfo_from_bc
#undef FCNAME
#define FCNAME MPIDI_QUOTE(FUNCNAME)
int MPIDU_Sctp_get_conninfo_from_bc( const char *bc, 
				     char *host_description, int maxlen,
				     int *port, MPIDU_Sock_ifaddr_t *ifaddr, 
				     int *hasIfaddr )
{
    int mpi_errno = MPI_SUCCESS;
    int str_errno;
#if !defined(HAVE_WINDOWS_H) && defined(HAVE_INET_PTON)
    char ifname[256];
#endif
    
    str_errno = MPIU_Str_get_string_arg(bc, MPIDI_CH3I_HOST_DESCRIPTION_KEY, 
				 host_description, maxlen);
    if (str_errno != MPIU_STR_SUCCESS) {
	/* --BEGIN ERROR HANDLING */
	if (str_errno == MPIU_STR_FAIL) {
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER,"**argstr_missinghost");
	}
	else {
	    /* MPIU_STR_TRUNCATED or MPIU_STR_NONEM */
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**argstr_hostd");
	}
	/* --END ERROR HANDLING-- */
    }
    str_errno = MPIU_Str_get_int_arg(bc, MPIDI_CH3I_PORT_KEY, port);
    if (str_errno != MPIU_STR_SUCCESS) {
	/* --BEGIN ERROR HANDLING */
	if (str_errno == MPIU_STR_FAIL) {
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**argstr_missingport");
	}
	else {
	    /* MPIU_STR_TRUNCATED or MPIU_STR_NONEM */
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER, "**argstr_port");
	}
	/* --END ERROR HANDLING-- */
    }
    /* ifname is optional */
    /* 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.  Note also that Windows does not have the inet_pton 
       routine; the Windows version of this routine will need to 
       be identified or written.  See also channels/sock/ch3_progress.c and
       channels/ssm/ch3_progress_connect.c */
    *hasIfaddr = 0;
#if !defined(HAVE_WINDOWS_H) && defined(HAVE_INET_PTON)
    str_errno = MPIU_Str_get_string_arg(bc, MPIDI_CH3I_IFNAME_KEY, 
					ifname, sizeof(ifname) );
    if (str_errno == MPIU_STR_SUCCESS) {
	/* Convert ifname into 4-byte ip address */
	/* Use AF_INET6 for IPv6 (inet_pton may still be used).
	   An address with more than 3 :'s is an IPv6 address */
	
	int rc = inet_pton( AF_INET, (const char *)ifname, ifaddr->ifaddr );
	if (rc == 0) {
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER,"**ifnameinvalid");
	}
	else if (rc < 0) {
	    /* af_inet not supported */
	    MPIU_ERR_SETANDJUMP(mpi_errno,MPI_ERR_OTHER,"**afinetinvalid");
	}
	else {
	    /* Success */
	    *hasIfaddr = 1;
	    ifaddr->len = 4;  /* IPv4 address */
	    ifaddr->type = AF_INET;
	}
    }
#endif
    
 fn_exit:
    
    return mpi_errno;
 fn_fail:
    goto fn_exit;
}